Update OPAL and ORTE for thread safety

(I swear, if I look this over one more time, I'll puke)

Signed-off-by: Ralph Castain <[email protected]>

Author: Ralph Castain

.gitignore

@@ -475,6 +475,7 @@ orte/test/system/opal_db
 orte/test/system/ulfm
 orte/test/system/pmixtool
 orte/test/system/orte_notify
+orte/test/system/threads
 
 orte/tools/orte-checkpoint/orte-checkpoint
 orte/tools/orte-checkpoint/orte-checkpoint.1

opal/mca/pmix/pmix2x/pmix2x.c

@@ -31,6 +31,7 @@
 #include "opal/mca/hwloc/base/base.h"
 #include "opal/runtime/opal.h"
 #include "opal/runtime/opal_progress_threads.h"
+#include "opal/threads/threads.h"
 #include "opal/util/argv.h"
 #include "opal/util/error.h"
 #include "opal/util/output.h"
@@ -164,6 +165,7 @@ static void return_local_event_hdlr(int status, opal_list_t *results,
     pmix_status_t pstatus;
     size_t n;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     if (NULL != cd->pmixcbfunc) {
         op = OBJ_NEW(pmix2x_opcaddy_t);
 
@@ -203,6 +205,8 @@ static void _event_hdlr(int sd, short args, void *cbdata)
     pmix2x_threadshift_t *cd = (pmix2x_threadshift_t*)cbdata;
     opal_pmix2x_event_t *event;
 
+    OPAL_ACQUIRE_OBJECT(cd);
+
     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
                         "%s _EVENT_HDLR RECEIVED NOTIFICATION FOR HANDLER %d OF STATUS %d",
                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME), (int)cd->id, cd->status);
@@ -312,6 +316,7 @@ void pmix2x_event_hdlr(size_t evhdlr_registration_id,
     /* now push it into the local thread */
     opal_event_assign(&cd->ev, opal_pmix_base.evbase,
                       -1, EV_WRITE, _event_hdlr, cd);
+    OPAL_POST_OBJECT(cd);
     opal_event_active(&cd->ev, EV_WRITE, 1);
 }
 
@@ -986,6 +991,7 @@ static void errreg_cbfunc (pmix_status_t status,
 {
     pmix2x_opcaddy_t *op = (pmix2x_opcaddy_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(op);
     op->event->index = errhandler_ref;
     opal_output_verbose(5, opal_pmix_base_framework.framework_output,
                         "PMIX2x errreg_cbfunc - error handler registered status=%d, reference=%lu",
@@ -1003,6 +1009,7 @@ static void _reg_hdlr(int sd, short args, void *cbdata)
     opal_value_t *kv;
     size_t n;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
                         "%s REGISTER HANDLER CODES %s",
                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
@@ -1067,6 +1074,7 @@ static void _dereg_hdlr(int sd, short args, void *cbdata)
     pmix2x_threadshift_t *cd = (pmix2x_threadshift_t*)cbdata;
     opal_pmix2x_event_t *event;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     /* look for this event */
     OPAL_LIST_FOREACH(event, &mca_pmix_pmix2x_component.events, opal_pmix2x_event_t) {
         if (cd->handler == event->index) {
@@ -1116,6 +1124,8 @@ static void _notify(int sd, short args, void *cbdata)
     pmix_data_range_t prange;
     opal_pmix2x_jobid_trkr_t *job, *jptr;
 
+    OPAL_ACQUIRE_OBJECT(cd);
+
     op = OBJ_NEW(pmix2x_opcaddy_t);
 
     /* convert the status */
@@ -1204,6 +1214,8 @@ static void infocbfunc(pmix_status_t status,
     opal_value_t *iptr;
     size_t n;
 
+    OPAL_ACQUIRE_OBJECT(cd);
+
     /* convert the array of pmix_info_t to the list of info */
     if (NULL != info) {
         results = OBJ_NEW(opal_list_t);
@@ -1294,6 +1306,8 @@ static void opcbfunc(pmix_status_t status, void *cbdata)
 {
     pmix2x_opcaddy_t *op = (pmix2x_opcaddy_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(op);
+
     if (NULL != op->opcbfunc) {
         op->opcbfunc(pmix2x_convert_rc(status), op->cbdata);
     }

opal/mca/pmix/pmix2x/pmix2x.h

@@ -156,6 +156,7 @@ OBJ_CLASS_DECLARATION(pmix2x_threadshift_t);
         _cd->cbdata = (cd);                                 \
         opal_event_assign(&((_cd)->ev), opal_pmix_base.evbase,   \
                           -1, EV_WRITE, (fn), (_cd));            \
+        OPAL_POST_OBJECT(_cd);                              \
         opal_event_active(&((_cd)->ev), EV_WRITE, 1);            \
     } while(0)
 
@@ -170,6 +171,7 @@ OBJ_CLASS_DECLARATION(pmix2x_threadshift_t);
         _cd->cbdata = (cd);                                 \
         opal_event_assign(&((_cd)->ev), opal_pmix_base.evbase,  \
                           -1, EV_WRITE, (fn), (_cd));           \
+        OPAL_POST_OBJECT(_cd);                              \
         opal_event_active(&((_cd)->ev), EV_WRITE, 1);           \
     } while(0)
 
@@ -185,6 +187,7 @@ OBJ_CLASS_DECLARATION(pmix2x_threadshift_t);
         _cd->cbdata = (cd);                                     \
         opal_event_assign(&((_cd)->ev), opal_pmix_base.evbase,  \
                      -1, EV_WRITE, (fn), (_cd));                \
+        OPAL_POST_OBJECT(_cd);                                  \
         opal_event_active(&((_cd)->ev), EV_WRITE, 1);           \
     } while(0)
 

opal/mca/pmix/pmix2x/pmix2x_client.c

@@ -27,6 +27,7 @@
 #endif
 
 #include "opal/hash_string.h"
+#include "opal/threads/threads.h"
 #include "opal/util/argv.h"
 #include "opal/util/proc.h"
 
@@ -44,6 +45,7 @@ static bool initialized = false;
         while ((a)) {                           \
             usleep(10);                         \
         }                                       \
+        OPAL_ACQUIRE_OBJECT(a);                 \
     } while (0)
 
 
@@ -53,11 +55,14 @@ static void errreg_cbfunc (pmix_status_t status,
 {
     opal_pmix2x_event_t *event = (opal_pmix2x_event_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(event);
+
     event->index = errhandler_ref;
     opal_output_verbose(5, opal_pmix_base_framework.framework_output,
                         "PMIX client errreg_cbfunc - error handler registered status=%d, reference=%lu",
                         status, (unsigned long)errhandler_ref);
     regactive = false;
+    OPAL_POST_OBJECT(regactive);
 }
 
 int pmix2x_client_init(opal_list_t *ilist)
@@ -272,6 +277,7 @@ static void opcbfunc(pmix_status_t status, void *cbdata)
 {
     pmix2x_opcaddy_t *op = (pmix2x_opcaddy_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(op);
     if (NULL != op->opcbfunc) {
         op->opcbfunc(pmix2x_convert_rc(status), op->cbdata);
     }
@@ -521,6 +527,8 @@ static void val_cbfunc(pmix_status_t status,
     int rc;
     opal_value_t val, *v=NULL;
 
+    OPAL_ACQUIRE_OBJECT(op);
+
     rc = pmix2x_convert_opalrc(status);
     if (PMIX_SUCCESS == status && NULL != kv) {
         rc = pmix2x_value_unload(&val, kv);
@@ -768,6 +776,8 @@ static void lk_cbfunc(pmix_status_t status,
     size_t n;
     opal_pmix2x_jobid_trkr_t *job, *jptr;
 
+    OPAL_ACQUIRE_OBJECT(op);
+
     /* this is in the PMIx local thread - need to threadshift to
      * our own thread as we will be accessing framework-global
      * lists and objects */
@@ -817,7 +827,7 @@ static void lk_cbfunc(pmix_status_t status,
         }
         r = &results;
     }
-release:
+  release:
     /* execute the callback */
     op->lkcbfunc(rc, r, op->cbdata);
 
@@ -994,6 +1004,8 @@ static void spcbfunc(pmix_status_t status,
     opal_jobid_t jobid=OPAL_JOBID_INVALID;
     opal_pmix2x_jobid_trkr_t *job;
 
+    OPAL_ACQUIRE_OBJECT(op);
+
     /* this is in the PMIx local thread - need to threadshift to
      * our own thread as we will be accessing framework-global
      * lists and objects */

opal/mca/pmix/pmix2x/pmix2x_server_north.c

@@ -29,6 +29,7 @@
 #include "opal/mca/hwloc/base/base.h"
 #include "opal/runtime/opal.h"
 #include "opal/runtime/opal_progress_threads.h"
+#include "opal/threads/threads.h"
 #include "opal/util/argv.h"
 #include "opal/util/error.h"
 #include "opal/util/output.h"
@@ -142,6 +143,7 @@ static void opal_opcbfunc(int status, void *cbdata)
 {
     pmix2x_opalcaddy_t *opalcaddy = (pmix2x_opalcaddy_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(opalcaddy);
     if (NULL != opalcaddy->opcbfunc) {
         opalcaddy->opcbfunc(pmix2x_convert_opalrc(status), opalcaddy->cbdata);
     }

opal/mca/pmix/pmix2x/pmix2x_server_south.c

@@ -32,6 +32,7 @@
 #include "opal/mca/hwloc/base/base.h"
 #include "opal/runtime/opal.h"
 #include "opal/runtime/opal_progress_threads.h"
+#include "opal/threads/threads.h"
 #include "opal/util/argv.h"
 #include "opal/util/error.h"
 #include "opal/util/output.h"
@@ -58,6 +59,7 @@ static size_t errhdler_ref = 0;
         while ((a)) {                           \
             usleep(10);                         \
         }                                       \
+        OPAL_ACQUIRE_OBJECT(a);                 \
     } while (0)
 
 static void errreg_cbfunc (pmix_status_t status,
@@ -66,22 +68,27 @@ static void errreg_cbfunc (pmix_status_t status,
 {
     volatile bool *active = (volatile bool*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(active);
     errhdler_ref = errhandler_ref;
     opal_output_verbose(5, opal_pmix_base_framework.framework_output,
                         "PMIX server errreg_cbfunc - error handler registered status=%d, reference=%lu",
                          status, (unsigned long)errhandler_ref);
+    OPAL_POST_OBJECT(active);
     *active = false;
 }
 
 static void opcbfunc(pmix_status_t status, void *cbdata)
 {
     pmix2x_opcaddy_t *op = (pmix2x_opcaddy_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(op);
+
     if (NULL != op->opcbfunc) {
         op->opcbfunc(pmix2x_convert_rc(status), op->cbdata);
     }
     if (op->active) {
         op->status = status;
+        OPAL_POST_OBJECT(op);
         op->active = false;
     } else {
         OBJ_RELEASE(op);
@@ -92,6 +99,7 @@ static void op2cbfunc(pmix_status_t status, void *cbdata)
 {
     volatile bool *active = (volatile bool*)cbdata;
 
+    OPAL_POST_OBJECT(active);
     *active = false;
 }
 
@@ -165,6 +173,7 @@ int pmix2x_server_init(opal_pmix_server_module_t *module,
 static void fincb(pmix_status_t status, void *cbdata)
 {
     volatile bool *active = (volatile bool*)cbdata;
+    OPAL_POST_OBJECT(active);
     *active = false;
 }
 
@@ -211,6 +220,8 @@ static void _reg_nspace(int sd, short args, void *cbdata)
     opal_pmix2x_jobid_trkr_t *job;
     pmix2x_opcaddy_t op;
 
+    OPAL_ACQUIRE_OBJECT(cd);
+
     /* we must threadshift this request as we might not be in an event
      * and we are going to access framework-global lists/objects */
 
@@ -301,6 +312,7 @@ int pmix2x_server_register_nspace(opal_jobid_t jobid,
     } else {
         opal_event_assign(&cd->ev, opal_pmix_base.evbase,
                           -1, EV_WRITE, _reg_nspace, cd);
+        OPAL_POST_OBJECT(cd);
         opal_event_active(&cd->ev, EV_WRITE, 1);
     }
 
@@ -311,10 +323,12 @@ static void tdcbfunc(pmix_status_t status, void *cbdata)
 {
     pmix2x_threadshift_t *cd = (pmix2x_threadshift_t*)cbdata;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     if (NULL != cd->opcbfunc) {
         cd->opcbfunc(pmix2x_convert_rc(status), cd->cbdata);
     }
     if (cd->active) {
+        OPAL_POST_OBJECT(cd);
         cd->active = false;
     } else {
         OBJ_RELEASE(cd);
@@ -326,6 +340,7 @@ static void _dereg_nspace(int sd, short args, void *cbdata)
     pmix2x_threadshift_t *cd = (pmix2x_threadshift_t*)cbdata;
     opal_pmix2x_jobid_trkr_t *jptr;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     /* if we don't already have it, we can ignore this */
     OPAL_LIST_FOREACH(jptr, &mca_pmix_pmix2x_component.jobids, opal_pmix2x_jobid_trkr_t) {
         if (jptr->jobid == cd->jobid) {
@@ -361,6 +376,7 @@ void pmix2x_server_deregister_nspace(opal_jobid_t jobid,
     } else {
         opal_event_assign(&cd->ev, opal_pmix_base.evbase,
                      -1, EV_WRITE, _dereg_nspace, cd);
+        OPAL_POST_OBJECT(cd);
         opal_event_active(&cd->ev, EV_WRITE, 1);
     }
 }
@@ -397,6 +413,7 @@ static void _dereg_client(int sd, short args, void *cbdata)
     opal_pmix2x_jobid_trkr_t *jptr;
     pmix_proc_t p;
 
+    OPAL_ACQUIRE_OBJECT(cd);
     /* if we don't already have it, we can ignore this */
     OPAL_LIST_FOREACH(jptr, &mca_pmix_pmix2x_component.jobids, opal_pmix2x_jobid_trkr_t) {
         if (jptr->jobid == cd->source->jobid) {
@@ -431,6 +448,7 @@ void pmix2x_server_deregister_client(const opal_process_name_t *proc,
     } else {
         opal_event_assign(&cd->ev, opal_pmix_base.evbase,
                      -1, EV_WRITE, _dereg_client, cd);
+        OPAL_POST_OBJECT(cd);
         opal_event_active(&cd->ev, EV_WRITE, 1);
     }
 }

opal/threads/threads.h

@@ -13,6 +13,7 @@
  * Copyright (c) 2010      Oracle and/or its affiliates.  All rights reserved.
  * Copyright (c) 2015-2017 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2017      Intel, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -114,6 +115,19 @@ OPAL_DECLSPEC OBJ_CLASS_DECLARATION(opal_thread_t);
         opal_condition_broadcast((cnd));    \
     } while(0);
 
+/* provide a macro for forward-proofing the shifting
+ * of objects between libevent threads - at some point, we
+ * may revamp that threading model */
+
+/* post an object to another thread - for now, we
+ * only have a memory barrier */
+#define OPAL_POST_OBJECT(o)     opal_atomic_wmb()
+
+/* acquire an object from another thread - for now,
+ * we only have a memory barrier */
+#define OPAL_ACQUIRE_OBJECT(o)  opal_atomic_rmb()
+
+
 
 OPAL_DECLSPEC int  opal_thread_start(opal_thread_t *);
 OPAL_DECLSPEC int  opal_thread_join(opal_thread_t *, void **thread_return);