Merge pull request #3662 from rhc54/topic/pmixupagain

Update to pmix v2.0.0rc1, including thread safety fixes

Author: Ralph Castain

opal/mca/pmix/pmix2x/pmix/VERSION

@@ -30,7 +30,7 @@ greek=
 # command, or with the date (if "git describe" fails) in the form of
 # "date<date>".
 
-repo_rev=git707f8cf
+repo_rev=git071ebc3
 
 # If tarball_version is not empty, it is used as the version string in
 # the tarball filename, regardless of all other versions listed in

opal/mca/pmix/pmix2x/pmix/src/atomics/sys/gcc_builtin/atomic.h

@@ -63,6 +63,8 @@ static inline void pmix_atomic_wmb(void)
 }
 
 #define PMIXMB() pmix_atomic_mb()
+#define PMIXRMB() pmix_atomic_rmb()
+#define PMIXWMB() pmix_atomic_wmb()
 
 /**********************************************************************
  *

opal/mca/pmix/pmix2x/pmix/src/atomics/sys/powerpc/atomic.h

@@ -10,7 +10,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2010      IBM Corporation.  All rights reserved.
+ * Copyright (c) 2010-2017 IBM Corporation.  All rights reserved.
  * Copyright (c) 2015-2016 Los Alamos National Security, LLC. All rights
  *                         reserved.
  * Copyright (c) 2017      Intel, Inc. All rights reserved.
@@ -30,10 +30,8 @@
 
 #define PMIXMB()  __asm__ __volatile__ ("sync" : : : "memory")
 #define PMIXRMB() __asm__ __volatile__ ("lwsync" : : : "memory")
-#define PMIXWMB() __asm__ __volatile__ ("eieio" : : : "memory")
+#define PMIXWMB() __asm__ __volatile__ ("lwsync" : : : "memory")
 #define PMIXISYNC() __asm__ __volatile__ ("isync" : : : "memory")
-#define PMIXSMP_SYNC  "sync \n\t"
-#define PMIXSMP_ISYNC "\n\tisync"
 
 
 /**********************************************************************

opal/mca/pmix/pmix2x/pmix/src/client/pmix_client.c

@@ -88,6 +88,7 @@ static const char pmix_version_string[] = PMIX_VERSION;
 static void _notify_complete(pmix_status_t status, void *cbdata)
 {
     pmix_event_chain_t *chain = (pmix_event_chain_t*)cbdata;
+    PMIX_ACQUIRE_OBJECT(chain);
     PMIX_RELEASE(chain);
 }
 
@@ -178,7 +179,7 @@ static void wait_cbfunc(struct pmix_peer_t *pr,
 
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix:client wait_cbfunc received");
-
+    PMIX_POST_OBJECT(active);
     *active = false;
 }
 
@@ -197,6 +198,7 @@ static void job_data(struct pmix_peer_t *pr,
     if (PMIX_SUCCESS != (rc = pmix_bfrop.unpack(buf, &nspace, &cnt, PMIX_STRING))) {
         PMIX_ERROR_LOG(rc);
         cb->status = PMIX_ERROR;
+        PMIX_POST_OBJECT(cb);
         cb->active = false;
         return;
     }
@@ -208,6 +210,7 @@ static void job_data(struct pmix_peer_t *pr,
     pmix_job_data_htable_store(pmix_globals.myid.nspace, buf);
 #endif
     cb->status = PMIX_SUCCESS;
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 
@@ -235,6 +238,7 @@ static void evhandler_reg_callbk(pmix_status_t status,
                                  void *cbdata)
 {
     volatile int *active = (volatile int*)cbdata;
+    PMIX_POST_OBJECT(active);
     *active = status;
 }
 
@@ -680,6 +684,9 @@ static void _putfn(int sd, short args, void *cbdata)
     uint8_t *tmp;
     size_t len;
 
+    /* need to acquire the cb object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cb);
+
     /* no need to push info that starts with "pmix" as that is
      * info we would have been provided at startup */
     if (0 == strncmp(cb->key, "pmix", 4)) {
@@ -757,6 +764,8 @@ static void _putfn(int sd, short args, void *cbdata)
         PMIX_RELEASE(kv);  // maintain accounting
     }
     cb->pstatus = rc;
+    /* post the data so the receiving thread can acquire it */
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 
@@ -802,6 +811,9 @@ static void _commitfn(int sd, short args, void *cbdata)
     pmix_buffer_t *msgout;
     pmix_cmd_t cmd=PMIX_COMMIT_CMD;
 
+    /* need to acquire the cb object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cb);
+
     msgout = PMIX_NEW(pmix_buffer_t);
     /* pack the cmd */
     if (PMIX_SUCCESS != (rc = pmix_bfrop.pack(msgout, &cmd, 1, PMIX_CMD))) {
@@ -850,6 +862,8 @@ static void _commitfn(int sd, short args, void *cbdata)
 
   done:
     cb->pstatus = rc;
+    /* post the data so the receiving thread can acquire it */
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
  }
 
@@ -901,6 +915,9 @@ static void _peersfn(int sd, short args, void *cbdata)
 #endif
     size_t i;
 
+    /* need to acquire the cb object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cb);
+
     /* cycle across our known nspaces */
     tmp = NULL;
 #if defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1)
@@ -955,6 +972,8 @@ static void _peersfn(int sd, short args, void *cbdata)
 
     done:
     cb->pstatus = rc;
+    /* post the data so the receiving thread can acquire it */
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 
@@ -1004,6 +1023,9 @@ static void _nodesfn(int sd, short args, void *cbdata)
     pmix_nspace_t *nsptr;
     pmix_nrec_t *nptr;
 
+    /* need to acquire the cb object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cb);
+
     /* cycle across our known nspaces */
     tmp = NULL;
     PMIX_LIST_FOREACH(nsptr, &pmix_globals.nspaces, pmix_nspace_t) {
@@ -1023,6 +1045,8 @@ static void _nodesfn(int sd, short args, void *cbdata)
     }
 
     cb->pstatus = rc;
+    /* post the data so the receiving thread can acquire it */
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 

opal/mca/pmix/pmix2x/pmix/src/client/pmix_client_connect.c

@@ -1,6 +1,6 @@
 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2014-2016 Intel, Inc.  All rights reserved.
+ * Copyright (c) 2014-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2014-2017 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2014      Artem Y. Polyakov <[email protected]>.
@@ -51,6 +51,8 @@
 #include "src/util/argv.h"
 #include "src/util/error.h"
 #include "src/util/output.h"
+#include "src/threads/threads.h"
+
 #include "src/mca/ptl/ptl.h"
 
 #include "pmix_client_ops.h"
@@ -344,5 +346,6 @@ static void op_cbfunc(pmix_status_t status, void *cbdata)
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
 
     cb->status = status;
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }

opal/mca/pmix/pmix2x/pmix/src/client/pmix_client_get.c

@@ -53,6 +53,7 @@
 
 #include "src/class/pmix_list.h"
 #include "src/buffer_ops/buffer_ops.h"
+#include "src/threads/threads.h"
 #include "src/util/argv.h"
 #include "src/util/compress.h"
 #include "src/util/error.h"
@@ -186,12 +187,14 @@ static void _value_cbfunc(pmix_status_t status, pmix_value_t *kv, void *cbdata)
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
     pmix_status_t rc;
 
+    PMIX_ACQUIRE_OBJECT(cb);
     cb->status = status;
     if (PMIX_SUCCESS == status) {
         if (PMIX_SUCCESS != (rc = pmix_bfrop.copy((void**)&cb->value, kv, PMIX_VALUE))) {
             PMIX_ERROR_LOG(rc);
         }
     }
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 
@@ -238,12 +241,12 @@ static pmix_buffer_t* _pack_get(char *nspace, pmix_rank_t rank,
     return msg;
 }
 
-/* this callback is coming from the usock recv, and thus
+/* this callback is coming from the ptl recv, and thus
  * is occurring inside of our progress thread - hence, no
  * need to thread shift */
 static void _getnb_cbfunc(struct pmix_peer_t *pr,
                           pmix_ptl_hdr_t *hdr,
-                         pmix_buffer_t *buf, void *cbdata)
+                          pmix_buffer_t *buf, void *cbdata)
 {
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
     pmix_cb_t *cb2;
@@ -486,6 +489,9 @@ static void _getnbfn(int fd, short flags, void *cbdata)
     char *tmp;
     bool my_nspace = false, my_rank = false;
 
+    /* cb was passed to us from another thread - acquire it */
+    PMIX_ACQUIRE_OBJECT(cb);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix: getnbfn value for proc %s:%d key %s",
                         cb->nspace, cb->rank,
@@ -739,11 +745,12 @@ static void _getnbfn(int fd, short flags, void *cbdata)
         rc = PMIX_ERROR;
         goto respond;
     }
-
+    /* we made a lot of changes to cb, so ensure they get
+     * written out before we return */
+    PMIX_POST_OBJECT(cb);
     return;
 
-respond:
-
+  respond:
     /* if a callback was provided, execute it */
     if (NULL != cb->value_cbfunc) {
         if (NULL != val)  {

opal/mca/pmix/pmix2x/pmix/src/client/pmix_client_pub.c

@@ -1,6 +1,6 @@
 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2014-2016 Intel, Inc.  All rights reserved.
+ * Copyright (c) 2014-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2014-2015 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2014      Artem Y. Polyakov <[email protected]>.
@@ -48,6 +48,7 @@
 
 #include "src/class/pmix_list.h"
 #include "src/buffer_ops/buffer_ops.h"
+#include "src/threads/threads.h"
 #include "src/util/argv.h"
 #include "src/util/error.h"
 #include "src/util/output.h"
@@ -304,7 +305,8 @@ PMIX_EXPORT pmix_status_t PMIx_Lookup_nb(char **keys,
 }
 
 PMIX_EXPORT pmix_status_t PMIx_Unpublish(char **keys,
-                               const pmix_info_t info[], size_t ninfo)
+                                         const pmix_info_t info[],
+                                         size_t ninfo)
 {
     pmix_status_t rc;
     pmix_cb_t *cb;
@@ -417,6 +419,8 @@ static void wait_cbfunc(struct pmix_peer_t *pr,
     int ret;
     int32_t cnt;
 
+    PMIX_ACQUIRE_OBJECT(cb);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix:client recv callback activated with %d bytes",
                         (NULL == buf) ? -1 : (int)buf->bytes_used);
@@ -437,6 +441,7 @@ static void op_cbfunc(pmix_status_t status, void *cbdata)
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
 
     cb->status = status;
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }
 
@@ -450,6 +455,8 @@ static void wait_lookup_cbfunc(struct pmix_peer_t *pr,
     pmix_pdata_t *pdata;
     size_t ndata;
 
+    PMIX_ACQUIRE_OBJECT(cb);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix:client recv callback activated with %d bytes",
                         (NULL == buf) ? -1 : (int)buf->bytes_used);
@@ -514,6 +521,7 @@ static void lookup_cbfunc(pmix_status_t status, pmix_pdata_t pdata[], size_t nda
     pmix_pdata_t *tgt = (pmix_pdata_t*)cb->cbdata;
     size_t i, j;
 
+    PMIX_ACQUIRE_OBJECT(cb);
     cb->status = status;
     if (PMIX_SUCCESS == status) {
         /* find the matching key in the provided info array - error if not found */
@@ -530,6 +538,6 @@ static void lookup_cbfunc(pmix_status_t status, pmix_pdata_t pdata[], size_t nda
             }
         }
     }
-
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }

opal/mca/pmix/pmix2x/pmix/src/client/pmix_client_spawn.c

@@ -1,6 +1,6 @@
 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2014-2016 Intel, Inc.  All rights reserved.
+ * Copyright (c) 2014-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2014-2017 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2014      Artem Y. Polyakov <[email protected]>.
@@ -48,6 +48,7 @@
 
 #include "src/class/pmix_list.h"
 #include "src/buffer_ops/buffer_ops.h"
+#include "src/threads/threads.h"
 #include "src/util/argv.h"
 #include "src/util/error.h"
 #include "src/util/output.h"
@@ -189,6 +190,8 @@ static void wait_cbfunc(struct pmix_peer_t *pr,
     pmix_status_t rc, ret;
     int32_t cnt;
 
+    PMIX_ACQUIRE_OBJECT(cb);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix:client recv callback activated with %d bytes",
                         (NULL == buf) ? -1 : (int)buf->bytes_used);
@@ -233,9 +236,11 @@ static void spawn_cbfunc(pmix_status_t status, char nspace[], void *cbdata)
 {
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
 
+    PMIX_ACQUIRE_OBJECT(cb);
     cb->status = status;
     if (NULL != nspace) {
         (void)strncpy(cb->nspace, nspace, PMIX_MAX_NSLEN);
     }
+    PMIX_POST_OBJECT(cb);
     cb->active = false;
 }

opal/mca/pmix/pmix2x/pmix/src/event/pmix_event_notification.c

@@ -254,6 +254,9 @@ static void progress_local_event_hdlr(pmix_status_t status,
                                       pmix_op_cbfunc_t cbfunc, void *thiscbdata,
                                       void *notification_cbdata)
 {
+    /* this may be in the host's thread, so we need to threadshift it
+     * before accessing our internal data */
+
     pmix_event_chain_t *chain = (pmix_event_chain_t*)notification_cbdata;
     size_t n, nsave, cnt;
     pmix_info_t *newinfo;
@@ -768,6 +771,9 @@ static void _notify_client_event(int sd, short args, void *cbdata)
     size_t n;
     bool matched, holdcd;
 
+    /* need to acquire the object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cd);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix_server: _notify_error notifying clients of error %s",
                         PMIx_Error_string(cd->status));
@@ -1056,6 +1062,9 @@ void pmix_event_timeout_cb(int fd, short flags, void *arg)
 {
     pmix_event_chain_t *ch = (pmix_event_chain_t*)arg;
 
+    /* need to acquire the object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(ch);
+
     ch->timer_active = false;
 
     /* remove it from the list */

opal/mca/pmix/pmix2x/pmix/src/event/pmix_event_registration.c

@@ -344,6 +344,9 @@ static void reg_event_hdlr(int sd, short args, void *cbdata)
     pmix_notify_caddy_t *ncd;
     pmix_event_chain_t *chain;
 
+    /* need to acquire the object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cd);
+
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix: register event_hdlr with %d infos", (int)cd->ninfo);
 
@@ -775,6 +778,9 @@ static void dereg_event_hdlr(int sd, short args, void *cbdata)
     size_t n;
     pmix_active_code_t *active;
 
+    /* need to acquire the object from its originating thread */
+    PMIX_ACQUIRE_OBJECT(cd);
+
     /* if I am not the server, then I need to notify the server
      * to remove my registration */
     if (PMIX_PROC_SERVER != pmix_globals.proc_type) {