Merge pull request #4111 from rhc54/topic/multiconnect

Cleanup some issues in connect/accept support across jobs started by …

Author: Ralph Castain

orte/orted/pmix/pmix_server_dyn.c

@@ -399,7 +399,11 @@ static void _cnlk(int status, opal_list_t *data, void *cbdata)
 
     /* restart the cnct processor */
     ORTE_PMIX_OPERATION(cd->procs, cd->info, _cnct, cd->cbfunc, cd->cbdata);
+    /* protect the re-referenced data */
+    cd->procs = NULL;
+    cd->info = NULL;
     OBJ_RELEASE(cd);
+    return;
 
   release:
     if (NULL != cd->cbfunc) {
@@ -415,6 +419,7 @@ static void _cnct(int sd, short args, void *cbdata)
     char **keys = NULL, *key;
     orte_job_t *jdata;
     int rc = ORTE_SUCCESS;
+    opal_value_t *kv;
 
     ORTE_ACQUIRE_OBJECT(cd);
 
@@ -444,6 +449,12 @@ static void _cnct(int sd, short args, void *cbdata)
             orte_util_convert_jobid_to_string(&key, nm->name.jobid);
             opal_argv_append_nosize(&keys, key);
             free(key);
+            /* we have to add the user's id to our list of info */
+            kv = OBJ_NEW(opal_value_t);
+            kv->key = strdup(OPAL_PMIX_USERID);
+            kv->type = OPAL_UINT32;
+            kv->data.uint32 = geteuid();
+            opal_list_append(cd->info, &kv->super);
             if (ORTE_SUCCESS != (rc = pmix_server_lookup_fn(&nm->name, keys, cd->info, _cnlk, cd))) {
                 opal_argv_free(keys);
                 goto release;

orte/orted/pmix/pmix_server_pub.c

@@ -394,6 +394,10 @@ int pmix_server_lookup_fn(opal_process_name_t *proc, char **keys,
             req->timeout = iptr->data.integer;
             continue;
         }
+        opal_output_verbose(2, orte_pmix_server_globals.output,
+                            "%s lookup directive %s for proc %s",
+                            ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), iptr->key,
+                            ORTE_NAME_PRINT(proc));
         if (OPAL_SUCCESS != (rc = opal_dss.pack(&req->msg, &iptr, 1, OPAL_VALUE))) {
             ORTE_ERROR_LOG(rc);
             OBJ_RELEASE(req);

orte/orted/pmix/pmix_server_register_fns.c

@@ -50,6 +50,8 @@
 #include "pmix_server_internal.h"
 #include "pmix_server.h"
 
+static void mycbfunc(int status, void *cbdata);
+
 /* stuff proc attributes for sending back to a proc */
 int orte_pmix_server_register_nspace(orte_job_t *jdata, bool force)
 {
@@ -472,5 +474,67 @@ int orte_pmix_server_register_nspace(orte_job_t *jdata, bool force)
                                           info, NULL, NULL);
     OPAL_LIST_RELEASE(info);
 
+    /* if the user has connected us to an external server, then we must
+     * assume there is going to be some cross-mpirun exchange, and so
+     * we protect against that situation by publishing the job info
+     * for this job - this allows any subsequent "connect" to retrieve
+     * the job info */
+    if (NULL != orte_data_server_uri) {
+        opal_buffer_t buf;
+
+        OBJ_CONSTRUCT(&buf, opal_buffer_t);
+        if (OPAL_SUCCESS != (rc = opal_dss.pack(&buf, &jdata, 1, ORTE_JOB))) {
+            ORTE_ERROR_LOG(rc);
+            OBJ_DESTRUCT(&buf);
+            return rc;
+        }
+        info = OBJ_NEW(opal_list_t);
+        /* create a key-value with the key being the string jobid
+         * and the value being the byte object */
+        kv = OBJ_NEW(opal_value_t);
+        orte_util_convert_jobid_to_string(&kv->key, jdata->jobid);
+        kv->type = OPAL_BYTE_OBJECT;
+        opal_dss.unload(&buf, (void**)&kv->data.bo.bytes, &kv->data.bo.size);
+        OBJ_DESTRUCT(&buf);
+        opal_list_append(info, &kv->super);
+
+        /* set the range to be session */
+        kv = OBJ_NEW(opal_value_t);
+        kv->key = strdup(OPAL_PMIX_RANGE);
+        kv->type = OPAL_UINT;
+        kv->data.uint = OPAL_PMIX_RANGE_SESSION;
+        opal_list_append(info, &kv->super);
+
+        /* set the persistence to be app */
+        kv = OBJ_NEW(opal_value_t);
+        kv->key = strdup(OPAL_PMIX_PERSISTENCE);
+        kv->type = OPAL_INT;
+        kv->data.integer = OPAL_PMIX_PERSIST_APP;
+        opal_list_append(info, &kv->super);
+
+        /* add our effective userid to the directives */
+        kv = OBJ_NEW(opal_value_t);
+        kv->key = strdup(OPAL_PMIX_USERID);
+        kv->type = OPAL_UINT32;
+        kv->data.uint32 = geteuid();
+        opal_list_append(info, &kv->super);
+
+        /* now publish it */
+        if (ORTE_SUCCESS != (rc = pmix_server_publish_fn(ORTE_PROC_MY_NAME,
+                                                         info, mycbfunc, info))) {
+            ORTE_ERROR_LOG(rc);
+        }
+    }
+
     return rc;
 }
+
+static void mycbfunc(int status, void *cbdata)
+{
+    opal_list_t *info = (opal_list_t*)cbdata;
+
+    if (ORTE_SUCCESS != status) {
+        ORTE_ERROR_LOG(status);
+    }
+    OPAL_LIST_RELEASE(info);
+}