Merge pull request #2493 from rhc54/topic/pmix

Update to latest PMIx master

Author: rhc54

opal/mca/pmix/pmix3x/pmix/VERSION

@@ -13,7 +13,7 @@
 # major, minor, and release are generally combined in the form
 # <major>.<minor>.<release>.
 
-major=3
+major=2
 minor=0
 release=0
 
@@ -30,7 +30,7 @@ greek=
 # command, or with the date (if "git describe" fails) in the form of
 # "date<date>".
 
-repo_rev=gitb041846
+repo_rev=git211a0ef
 
 # If tarball_version is not empty, it is used as the version string in
 # the tarball filename, regardless of all other versions listed in
@@ -44,7 +44,7 @@ tarball_version=
 
 # The date when this release was created
 
-date="Oct 27, 2016"
+date="Dec 01, 2016"
 
 # The shared library version of each of PMIx's public libraries.
 # These versions are maintained in accordance with the "Library

opal/mca/pmix/pmix3x/pmix/config/pmix.m4

@@ -926,4 +926,3 @@ AC_DEFUN([PMIX_DO_AM_CONDITIONALS],[
     ])
     pmix_did_am_conditionals=yes
 ])dnl
-

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

@@ -72,6 +72,7 @@
 #endif /* PMIX_ENABLE_DSTORE */
 
 #include "pmix_client_ops.h"
+#include "src/include/pmix_jobdata.h"
 
 #define PMIX_MAX_RETRIES 10
 
@@ -191,8 +192,11 @@ static void job_data(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
     }
     assert(NULL != nspace);
     free(nspace);
+
     /* decode it */
-    pmix_client_process_nspace_blob(pmix_globals.myid.nspace, buf);
+#if !(defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1))
+    pmix_job_data_htable_store(pmix_globals.myid.nspace, buf);
+#endif
     cb->status = PMIX_SUCCESS;
     cb->active = false;
 }
@@ -715,12 +719,27 @@ static void _peersfn(int sd, short args, void *cbdata)
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
     pmix_status_t rc;
     char **nsprocs=NULL, **nsps=NULL, **tmp;
+#if !(defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1))
     pmix_nspace_t *nsptr;
     pmix_nrec_t *nptr;
+#endif
     size_t i;
 
     /* cycle across our known nspaces */
     tmp = NULL;
+#if defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1)
+    if (PMIX_SUCCESS == (rc = pmix_dstore_fetch(cb->nspace, PMIX_RANK_WILDCARD,
+                      cb->key, &cb->value))) {
+
+        tmp = pmix_argv_split(cb->value->data.string, ',');
+        for (i=0; NULL != tmp[i]; i++) {
+            pmix_argv_append_nosize(&nsps, cb->nspace);
+            pmix_argv_append_nosize(&nsprocs, tmp[i]);
+        }
+        pmix_argv_free(tmp);
+        tmp = NULL;
+    }
+#else
     PMIX_LIST_FOREACH(nsptr, &pmix_globals.nspaces, pmix_nspace_t) {
         if (0 == strncmp(nsptr->nspace, cb->nspace, PMIX_MAX_NSLEN)) {
             /* cycle across the nodes in this nspace */
@@ -738,6 +757,7 @@ static void _peersfn(int sd, short args, void *cbdata)
             }
         }
     }
+#endif
     if (0 == (i = pmix_argv_count(nsps))) {
         /* we don't know this nspace */
         rc = PMIX_ERR_NOT_FOUND;
@@ -1010,160 +1030,8 @@ static pmix_status_t send_connect_ack(int sd)
     return PMIX_SUCCESS;
 }
 
-void pmix_client_process_nspace_blob(const char *nspace, pmix_buffer_t *bptr)
+static pmix_status_t usock_connect(struct sockaddr *addr, int *fd)
 {
-    pmix_status_t rc;
-    int32_t cnt;
-    int rank;
-    pmix_kval_t *kptr, *kp2, kv;
-    pmix_buffer_t buf2;
-    pmix_byte_object_t *bo;
-    size_t nnodes, i, j;
-    pmix_nspace_t *nsptr, *nsptr2;
-    pmix_nrec_t *nrec, *nr2;
-    char **procs;
-
-    pmix_output_verbose(2, pmix_globals.debug_output,
-                        "pmix: PROCESSING BLOB FOR NSPACE %s", nspace);
-
-    /* cycle across our known nspaces */
-    nsptr = NULL;
-    PMIX_LIST_FOREACH(nsptr2, &pmix_globals.nspaces, pmix_nspace_t) {
-        if (0 == strcmp(nsptr2->nspace, nspace)) {
-            nsptr = nsptr2;
-            break;
-        }
-    }
-    if (NULL == nsptr) {
-        /* we don't know this nspace - add it */
-        nsptr = PMIX_NEW(pmix_nspace_t);
-        (void)strncpy(nsptr->nspace, nspace, PMIX_MAX_NSLEN);
-        pmix_list_append(&pmix_globals.nspaces, &nsptr->super);
-    }
-
-    /* unpack any info structs provided */
-    cnt = 1;
-    kptr = PMIX_NEW(pmix_kval_t);
-    while (PMIX_SUCCESS == (rc = pmix_bfrop.unpack(bptr, kptr, &cnt, PMIX_KVAL))) {
-        if (0 == strcmp(kptr->key, PMIX_PROC_BLOB)) {
-            /* transfer the byte object for unpacking */
-            bo = &(kptr->value->data.bo);
-            PMIX_CONSTRUCT(&buf2, pmix_buffer_t);
-            PMIX_LOAD_BUFFER(&buf2, bo->bytes, bo->size);
-            /* start by unpacking the rank */
-            cnt = 1;
-            if (PMIX_SUCCESS != (rc = pmix_bfrop.unpack(&buf2, &rank, &cnt, PMIX_PROC_RANK))) {
-                PMIX_ERROR_LOG(rc);
-                PMIX_DESTRUCT(&buf2);
-                return;
-            }
-            kp2 = PMIX_NEW(pmix_kval_t);
-            kp2->key = strdup(PMIX_RANK);
-            PMIX_VALUE_CREATE(kp2->value, 1);
-            kp2->value->type = PMIX_PROC_RANK;
-            kp2->value->data.rank = rank;
-            if (PMIX_SUCCESS != (rc = pmix_hash_store(&nsptr->internal, rank, kp2))) {
-                PMIX_ERROR_LOG(rc);
-            }
-            PMIX_RELEASE(kp2); // maintain accounting
-            cnt = 1;
-            kp2 = PMIX_NEW(pmix_kval_t);
-            while (PMIX_SUCCESS == (rc = pmix_bfrop.unpack(&buf2, kp2, &cnt, PMIX_KVAL))) {
-                /* this is data provided by a job-level exchange, so store it
-                 * in the job-level data hash_table */
-                if (PMIX_SUCCESS != (rc = pmix_hash_store(&nsptr->internal, rank, kp2))) {
-                    PMIX_ERROR_LOG(rc);
-                }
-                PMIX_RELEASE(kp2); // maintain accounting
-                kp2 = PMIX_NEW(pmix_kval_t);
-            }
-            /* cleanup */
-            PMIX_DESTRUCT(&buf2);  // releases the original kptr data
-            PMIX_RELEASE(kp2);
-        } else if (0 == strcmp(kptr->key, PMIX_MAP_BLOB)) {
-            /* transfer the byte object for unpacking */
-            bo = &(kptr->value->data.bo);
-            PMIX_CONSTRUCT(&buf2, pmix_buffer_t);
-            PMIX_LOAD_BUFFER(&buf2, bo->bytes, bo->size);
-            /* start by unpacking the number of nodes */
-            cnt = 1;
-            if (PMIX_SUCCESS != (rc = pmix_bfrop.unpack(&buf2, &nnodes, &cnt, PMIX_SIZE))) {
-                PMIX_ERROR_LOG(rc);
-                PMIX_DESTRUCT(&buf2);
-                return;
-            }
-            /* unpack the list of procs on each node */
-            for (i=0; i < nnodes; i++) {
-                cnt = 1;
-                PMIX_CONSTRUCT(&kv, pmix_kval_t);
-                if (PMIX_SUCCESS != (rc = pmix_bfrop.unpack(&buf2, &kv, &cnt, PMIX_KVAL))) {
-                    PMIX_ERROR_LOG(rc);
-                    PMIX_DESTRUCT(&buf2);
-                    PMIX_DESTRUCT(&kv);
-                    return;
-                }
-                /* the name of the node is in the key, and the value is
-                 * a comma-delimited list of procs on that node. See if we already
-                 * have this node */
-                 nrec = NULL;
-                 PMIX_LIST_FOREACH(nr2, &nsptr->nodes, pmix_nrec_t) {
-                    if (0 == strcmp(nr2->name, kv.key)) {
-                        nrec = nr2;
-                        break;
-                    }
-                }
-                if (NULL == nrec) {
-                    /* Create a node record and store that list */
-                    nrec = PMIX_NEW(pmix_nrec_t);
-                    nrec->name = strdup(kv.key);
-                    pmix_list_append(&nsptr->nodes, &nrec->super);
-                } else {
-                    /* refresh the list */
-                    if (NULL != nrec->procs) {
-                        free(nrec->procs);
-                    }
-                }
-                nrec->procs = strdup(kv.value->data.string);
-                /* split the list of procs so we can store their
-                 * individual location data */
-                procs = pmix_argv_split(nrec->procs, ',');
-                for (j=0; NULL != procs[j]; j++) {
-                    /* store the hostname for each proc - again, this is
-                     * data obtained via a job-level exchange, so store it
-                     * in the job-level data hash_table */
-                     kp2 = PMIX_NEW(pmix_kval_t);
-                     kp2->key = strdup(PMIX_HOSTNAME);
-                     kp2->value = (pmix_value_t*)malloc(sizeof(pmix_value_t));
-                     kp2->value->type = PMIX_STRING;
-                     kp2->value->data.string = strdup(nrec->name);
-                     rank = strtol(procs[j], NULL, 10);
-                     if (PMIX_SUCCESS != (rc = pmix_hash_store(&nsptr->internal, rank, kp2))) {
-                        PMIX_ERROR_LOG(rc);
-                    }
-                    PMIX_RELEASE(kp2); // maintain accounting
-                }
-                pmix_argv_free(procs);
-                PMIX_DESTRUCT(&kv);
-            }
-            /* cleanup */
-            PMIX_DESTRUCT(&buf2);  // releases the original kptr data
-        } else {
-            /* this is job-level data, so just add it to that hash_table
-             * with the wildcard rank */
-            if (PMIX_SUCCESS != (rc = pmix_hash_store(&nsptr->internal, PMIX_RANK_WILDCARD, kptr))) {
-                PMIX_ERROR_LOG(rc);
-            }
-         }
-         PMIX_RELEASE(kptr);
-         kptr = PMIX_NEW(pmix_kval_t);
-         cnt = 1;
-     }
-    /* need to release the leftover kptr */
-     PMIX_RELEASE(kptr);
- }
-
- static pmix_status_t usock_connect(struct sockaddr *addr, int *fd)
- {
     int sd=-1;
     pmix_status_t rc;
     pmix_socklen_t addrlen = 0;

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

@@ -54,6 +54,7 @@
 #include "src/usock/usock.h"
 
 #include "pmix_client_ops.h"
+#include "src/include/pmix_jobdata.h"
 
 /* callback for wait completion */
 static void wait_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
@@ -313,7 +314,9 @@ static void wait_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
             continue;
         }
         /* extract and process any proc-related info for this nspace */
-        pmix_client_process_nspace_blob(nspace, bptr);
+#if !(defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1))
+        pmix_job_data_htable_store(nspace, bptr);
+#endif
         PMIX_RELEASE(bptr);
     }
     if (PMIX_ERR_UNPACK_READ_PAST_END_OF_BUFFER != rc) {

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

@@ -24,6 +24,7 @@
 #include <pmix_rename.h>
 
 #include "src/include/pmix_globals.h"
+#include "src/include/pmix_jobdata.h"
 
 #ifdef HAVE_STRING_H
 #include <string.h>
@@ -58,6 +59,7 @@
 #endif /* PMIX_ENABLE_DSTORE */
 
 #include "pmix_client_ops.h"
+#include "src/include/pmix_jobdata.h"
 
 static pmix_buffer_t* _pack_get(char *nspace, pmix_rank_t rank,
                                const pmix_info_t info[], size_t ninfo,
@@ -283,8 +285,12 @@ static void _getnb_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
         goto done;
     }
 
-#if (PMIX_ENABLE_DSTORE == 1)
-    rc = pmix_dstore_fetch(nptr->nspace, cb->rank, cb->key, &val);
+#if (defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1))
+    if (PMIX_SUCCESS != (rc = pmix_dstore_fetch(nptr->nspace, cb->rank, cb->key, &val))){
+        /* DO NOT error log this status - it is perfectly okay
+         * for a key not to be found */
+        goto done;
+    }
 #else
     /* we received the entire blob for this process, so
      * unpack and store it in the modex - this could consist
@@ -308,7 +314,12 @@ static void _getnb_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
                     return;
                 }
                 free(nspace);
-                pmix_client_process_nspace_blob(cb->nspace, bptr);
+                pmix_job_data_htable_store(cb->nspace, bptr);
+
+                /* Check if the key is in this blob */
+
+                pmix_hash_fetch(&nptr->internal, cb->rank, cb->key, &val);
+
             } else {
                 cnt = 1;
                 cur_kval = PMIX_NEW(pmix_kval_t);
@@ -550,11 +561,29 @@ static void _getnbfn(int fd, short flags, void *cbdata)
         return;
     }
 
+#if defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1)
+    if (PMIX_SUCCESS == (rc = pmix_hash_fetch(&nptr->internal, cb->rank, cb->key, &val))) {
+        /* found it - we are in an event, so we can
+         * just execute the callback */
+        cb->value_cbfunc(rc, val, cb->cbdata);
+        /* cleanup */
+        if (NULL != val) {
+            PMIX_VALUE_RELEASE(val);
+        }
+        PMIX_RELEASE(cb);
+        return;
+    }
+#endif
+
     /* if the key is in the PMIx namespace, then they are looking for data
      * that was provided at startup */
     if (0 == strncmp(cb->key, "pmix", 4)) {
         /* should be in the internal hash table. */
+#if defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1)
+        if (PMIX_SUCCESS == (rc = pmix_dstore_fetch(cb->nspace, cb->rank, cb->key, &val))) {
+#else
         if (PMIX_SUCCESS == (rc = pmix_hash_fetch(&nptr->internal, cb->rank, cb->key, &val))) {
+#endif
             /* found it - we are in an event, so we can
              * just execute the callback */
             cb->value_cbfunc(rc, val, cb->cbdata);

opal/mca/pmix/pmix3x/pmix/src/client/pmix_client_ops.h

@@ -26,9 +26,6 @@ typedef struct {
 
 extern pmix_client_globals_t pmix_client_globals;
 
-void pmix_client_process_nspace_blob(const char *nspace, pmix_buffer_t *bptr);
-
-
 END_C_DECLS
 
 #endif /* PMIX_CLIENT_OPS_H */

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

@@ -54,6 +54,7 @@
 #include "src/usock/usock.h"
 
 #include "pmix_client_ops.h"
+#include "src/include/pmix_jobdata.h"
 
 static void wait_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
                         pmix_buffer_t *buf, void *cbdata);
@@ -179,7 +180,7 @@ static void wait_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
 {
     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
     char nspace[PMIX_MAX_NSLEN+1];
-    char *n2;
+    char *n2 = NULL;
     pmix_status_t rc, ret;
     int32_t cnt;
 
@@ -203,10 +204,15 @@ static void wait_cbfunc(struct pmix_peer_t *pr, pmix_usock_hdr_t *hdr,
             PMIX_ERROR_LOG(rc);
             ret = rc;
         }
+        pmix_output_verbose(1, pmix_globals.debug_output,
+                        "pmix:client recv '%s'", n2);
+
         if (NULL != n2) {
             (void)strncpy(nspace, n2, PMIX_MAX_NSLEN);
+#if !(defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1))
             /* extract and process any proc-related info for this nspace */
-            pmix_client_process_nspace_blob(nspace, buf);
+            pmix_job_data_htable_store(nspace, buf);
+#endif
             free(n2);
         }
     }
@@ -226,4 +232,3 @@ static void spawn_cbfunc(pmix_status_t status, char nspace[], void *cbdata)
     }
     cb->active = false;
 }
-