Merge pull request open-mpi#609 from rhc54/cmr2.0/pmix

Sync to PMIx master

Author: hppritcha

ompi/dpm/dpm.c

@@ -139,7 +139,13 @@ int ompi_dpm_connect_accept(ompi_communicator_t *comm, int root,
         opal_argv_append_nosize(&members, nstring);
         free(nstring);
         /* have to add the number of procs in the job so the remote side
-         * can correctly add the procs by computing their names */
+         * can correctly add the procs by computing their names, and our nspace
+         * so they can update their records */
+        if (NULL == (nstring = (char*)opal_pmix.get_nspace(OMPI_PROC_MY_NAME->jobid))) {
+            opal_argv_free(members);
+            return OMPI_ERR_NOT_SUPPORTED;
+        }
+        opal_argv_append_nosize(&members, nstring);
         (void)asprintf(&nstring, "%d", size);
         opal_argv_append_nosize(&members, nstring);
         free(nstring);
@@ -171,6 +177,11 @@ int ompi_dpm_connect_accept(ompi_communicator_t *comm, int root,
             }
             opal_argv_append_nosize(&members, nstring);
             free(nstring);
+            if (NULL == (nstring = (char*)opal_pmix.get_nspace(proc_list[i]->super.proc_name.jobid))) {
+                opal_argv_free(members);
+                return OMPI_ERR_NOT_SUPPORTED;
+            }
+            opal_argv_append_nosize(&members, nstring);
         }
         if (!dense) {
             free(proc_list);
@@ -246,6 +257,17 @@ int ompi_dpm_connect_accept(ompi_communicator_t *comm, int root,
             OPAL_LIST_DESTRUCT(&mlist);
             goto exit;
         }
+        /* step over the nspace */
+        ++i;
+        if (NULL == members[i]) {
+            /* this shouldn't happen and is an error */
+            OMPI_ERROR_LOG(OMPI_ERR_BAD_PARAM);
+            OPAL_LIST_DESTRUCT(&mlist);
+            opal_argv_free(members);
+            free(rport);
+            rc = OMPI_ERR_BAD_PARAM;
+            goto exit;
+        }
         /* if the rank is wildcard, then we need to add all procs
          * in that job to the list */
         if (OPAL_VPID_WILDCARD == nm->name.vpid) {
@@ -295,6 +317,16 @@ int ompi_dpm_connect_accept(ompi_communicator_t *comm, int root,
             OPAL_LIST_DESTRUCT(&rlist);
             goto exit;
         }
+        /* next entry is the nspace - register it */
+        ++i;
+        if (NULL == members[i]) {
+            OMPI_ERROR_LOG(OMPI_ERR_NOT_SUPPORTED);
+            opal_argv_free(members);
+            OPAL_LIST_DESTRUCT(&ilist);
+            OPAL_LIST_DESTRUCT(&rlist);
+            goto exit;
+        }
+        opal_pmix.register_jobid(nm->name.jobid, members[i]);
         if (OPAL_VPID_WILDCARD == nm->name.vpid) {
             jobid = nm->name.jobid;
             OBJ_RELEASE(nm);

opal/mca/pmix/cray/pmix_cray.c

@@ -78,6 +78,9 @@ static int cray_unpublish_nb(char **keys, opal_list_t *info,
 static const char *cray_get_version(void);
 static int cray_store_local(const opal_process_name_t *proc,
                           opal_value_t *val);
+static const char *cray_get_nspace(opal_jobid_t jobid);
+static void cray_register_jobid(opal_jobid_t jobid, const char *nspace);
+
 #if 0
 static bool cray_get_attr(const char *attr, opal_value_t **kv);
 #endif
@@ -109,7 +112,9 @@ const opal_pmix_base_module_t opal_pmix_cray_module = {
     .get_version = cray_get_version,
     .register_errhandler = opal_pmix_base_register_handler,
     .deregister_errhandler = opal_pmix_base_deregister_handler,
-    .store_local = cray_store_local
+    .store_local = cray_store_local,
+    .get_nspace = cray_get_nspace,
+    .register_jobid = cray_register_jobid
 };
 
 // usage accounting
@@ -814,6 +819,16 @@ static int cray_store_local(const opal_process_name_t *proc,
     return OPAL_SUCCESS;
 }
 
+static const char *cray_get_nspace(opal_jobid_t jobid)
+{
+    return NULL;
+}
+
+static void cray_register_jobid(opal_jobid_t jobid, const char *nspace)
+{
+    return;
+}
+
 static char* pmix_error(int pmix_err)
 {
     char * err_msg;

opal/mca/pmix/pmix.h

@@ -62,9 +62,9 @@ extern int opal_pmix_base_exchange(opal_value_t *info,
                 OPAL_ERROR_LOG((r));                                     \
             }                                                            \
         }                                                                \
-        /* do not destruct the keyval as we don't own */                 \
-        /* the data - the caller will take care of the */                \
-        /* key and value storage, and the kv itself has none */          \
+        /* opal_value_load makes a copy of the data, so release it */    \
+        _kv.key = NULL;                                                  \
+        OBJ_DESTRUCT(&(_kv));                                            \
     } while(0);
 
 /**
@@ -701,6 +701,12 @@ typedef void (*opal_pmix_base_module_deregister_fn_t)(void);
 typedef int (*opal_pmix_base_module_store_fn_t)(const opal_process_name_t *proc,
                                                 opal_value_t *val);
 
+/* retrieve the nspace corresponding to a given jobid */
+typedef const char* (*opal_pmix_base_module_get_nspace_fn_t)(opal_jobid_t jobid);
+
+/* register a jobid-to-nspace pair */
+typedef void (*opal_pmix_base_module_register_jobid_fn_t)(opal_jobid_t jobid, const char *nspace);
+
 /*
  * the standard public API data structure
  */
@@ -745,6 +751,8 @@ typedef struct {
     opal_pmix_base_module_register_fn_t               register_errhandler;
     opal_pmix_base_module_deregister_fn_t             deregister_errhandler;
     opal_pmix_base_module_store_fn_t                  store_local;
+    opal_pmix_base_module_get_nspace_fn_t             get_nspace;
+    opal_pmix_base_module_register_jobid_fn_t         register_jobid;
 } opal_pmix_base_module_t;
 
 typedef struct {

opal/mca/pmix/pmix1xx/pmix/VERSION

@@ -30,7 +30,7 @@ greek=a1
 # command, or with the date (if "git describe" fails) in the form of
 # "date<date>".
 
-repo_rev=git89680d6
+repo_rev=git0a8e0d9
 
 # 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="Sep 10, 2015"
+date="Sep 23, 2015"
 
 # The shared library version of each of PMIx's public libraries.
 # These versions are maintained in accordance with the "Library

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

@@ -779,21 +779,28 @@ static pmix_status_t recv_connect_ack(int sd)
     pmix_status_t rc;
     struct timeval tv, save;
     pmix_socklen_t sz;
+    bool sockopt = true;
 
     pmix_output_verbose(2, pmix_globals.debug_output,
                         "pmix: RECV CONNECT ACK FROM SERVER");
 
     /* get the current timeout value so we can reset to it */
     sz = sizeof(save);
     if (0 != getsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, (void*)&save, &sz)) {
-        return PMIX_ERR_UNREACH;
-    }
-
-    /* set a timeout on the blocking recv so we don't hang */
-    tv.tv_sec  = 2;
-    tv.tv_usec = 0;
-    if (0 != setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))) {
-        return PMIX_ERR_UNREACH;
+        if (ENOPROTOOPT == errno) {
+            sockopt = false;
+        } else {
+            return PMIX_ERR_UNREACH;
+        }
+    } else {
+        /* set a timeout on the blocking recv so we don't hang */
+        tv.tv_sec  = 2;
+        tv.tv_usec = 0;
+        if (0 != setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))) {
+            pmix_output_verbose(2, pmix_globals.debug_output,
+                                "pmix: recv_connect_ack could not setsockopt SO_RCVTIMEO");
+            return PMIX_ERR_UNREACH;
+        }
     }
 
     /* receive the status reply */
@@ -825,9 +832,11 @@ static pmix_status_t recv_connect_ack(int sd)
         return rc;
     }
 
-    /* return the socket to normal */
-    if (0 != setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, &save, sz)) {
-        return PMIX_ERR_UNREACH;
+    if (sockopt) {
+        /* return the socket to normal */
+        if (0 != setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, &save, sz)) {
+            return PMIX_ERR_UNREACH;
+        }
     }
 
     return PMIX_SUCCESS;

opal/mca/pmix/pmix1xx/pmix/src/server/pmix_server.c

@@ -102,12 +102,19 @@ PMIX_CLASS_INSTANCE(pmix_usock_queue_t,
 static void scon(pmix_shift_caddy_t *p)
 {
     p->active = false;
+    p->kv = NULL;
     p->relfn = NULL;
     p->relcbd = NULL;
 }
+static void scdes(pmix_shift_caddy_t *p)
+{
+    if (NULL != p->kv) {
+        PMIX_RELEASE(p->kv);
+    }
+}
 PMIX_CLASS_INSTANCE(pmix_shift_caddy_t,
                     pmix_object_t,
-                    scon, NULL);
+                    scon, scdes);
 
 
  #define PMIX_THREADSHIFT(r, c)                       \

opal/mca/pmix/pmix1xx/pmix1.h

@@ -30,11 +30,24 @@
 
 BEGIN_C_DECLS
 
-OPAL_DECLSPEC extern opal_pmix_base_component_t mca_pmix_pmix1_component;
+typedef struct {
+  opal_pmix_base_component_t super;
+  opal_list_t jobids;
+  bool native_launch;
+} mca_pmix_pmix1_component_t;
+
+OPAL_DECLSPEC extern mca_pmix_pmix1_component_t mca_pmix_pmix1xx_component;
 
 OPAL_DECLSPEC extern const opal_pmix_base_module_t opal_pmix_pmix1xx_module;
 
 /****  INTERNAL OBJECTS  ****/
+typedef struct {
+    opal_list_item_t super;
+    opal_jobid_t jobid;
+    char nspace[PMIX_MAX_NSLEN + 1];
+} opal_pmix1_jobid_trkr_t;
+OBJ_CLASS_DECLARATION(opal_pmix1_jobid_trkr_t);
+
 typedef struct {
     opal_object_t super;
     pmix_proc_t p;