Enable use of OFI fabrics for launch and other collective operations. Update the PMIx repo to the latest master to get the required support for the server to "push" modex info, and to retrieve all its own "modex" values for sending back to mpirun. Have mpirun cache them in its local modex hash as OFI goes point-to-point direct and doesn't route - so the remote daemons don't need a copy of this connection info.

Remove the opal_ignore from the RML/OFI component, but disable that component unless the user specifically requests it via the "rml_ofi_desired=1" MCA param. This will let us test compile in various environments without interfering with operations while we continue to debug

Fix an error when computing the number of infos during server init

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

Author: Ralph Castain

opal/mca/pmix/pmix2x/pmix/include/pmix_common.h

@@ -124,6 +124,8 @@ typedef uint32_t pmix_rank_t;
 #define PMIX_CONNECT_SYSTEM_FIRST           "pmix.cnct.sys.first"   // (bool) Preferentially look for a system-level PMIx server first
 #define PMIX_REGISTER_NODATA                "pmix.reg.nodata"       // (bool) Registration is for nspace only, do not copy job data
 #define PMIX_SERVER_ENABLE_MONITORING       "pmix.srv.monitor"      // (bool) Enable PMIx internal monitoring by server
+#define PMIX_SERVER_NSPACE                  "pmix.srv.nspace"       // (char*) Name of the nspace to use for this server
+#define PMIX_SERVER_RANK                    "pmix.srv.rank"         // (pmix_rank_t) Rank of this server
 
 
 /* identification attributes */

opal/mca/pmix/pmix2x/pmix/src/buffer_ops/copy.c

@@ -425,7 +425,7 @@ PMIX_EXPORT pmix_status_t pmix_value_xfer(pmix_value_t *p, pmix_value_t *src)
             break;
         }
         /* allocate space and do the copy */
-        switch (src->type) {
+        switch (src->data.darray->type) {
             case PMIX_UINT8:
             case PMIX_INT8:
             case PMIX_BYTE:

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

@@ -111,7 +111,7 @@ PMIX_EXPORT pmix_status_t PMIx_Get(const pmix_proc_t *proc, const char key[],
     PMIX_RELEASE(cb);
 
     pmix_output_verbose(2, pmix_globals.debug_output,
-                        "pmix:client get completed");
+                        "pmix:client get completed %d", rc);
 
     return rc;
 }
@@ -464,7 +464,7 @@ static pmix_status_t process_val(pmix_value_t *val,
     }
     nvals = 0;
     for (n=0; n < nsize; n++) {
-        if (PMIX_SUCCESS != (rc = pmix_pointer_array_add(results, &info[n]))) {
+        if (0 > (rc = pmix_pointer_array_add(results, &info[n]))) {
             return rc;
         }
         ++nvals;
@@ -536,25 +536,45 @@ static void _getnbfn(int fd, short flags, void *cbdata)
         /* if the rank is WILDCARD, then they want all the job-level info,
          * so no need to check the modex */
         if (PMIX_RANK_WILDCARD != cb->rank) {
+            rc = PMIX_ERR_NOT_FOUND;
 #if defined(PMIX_ENABLE_DSTORE) && (PMIX_ENABLE_DSTORE == 1)
-            if (PMIX_SUCCESS == (rc = pmix_dstore_fetch(nptr->nspace, cb->rank, NULL, &val))) {
-#else
-            if (PMIX_SUCCESS == (rc = pmix_hash_fetch(&nptr->modex, cb->rank, NULL, &val))) {
+            /* my own data is in the hash table, so don't bother looking
+             * in the dstore if that is what they want */
+            if (pmix_globals.myid.rank != cb->rank) {
+                if (PMIX_SUCCESS == (rc = pmix_dstore_fetch(nptr->nspace, cb->rank, NULL, &val))) {
+                    pmix_output_verbose(2, pmix_globals.debug_output,
+                                        "pmix_get[%d]: value retrieved from dstore", __LINE__);
+                    if (PMIX_SUCCESS != (rc = process_val(val, &nvals, &results))) {
+                        cb->value_cbfunc(rc, NULL, cb->cbdata);
+                        /* cleanup */
+                        if (NULL != val) {
+                            PMIX_VALUE_RELEASE(val);
+                        }
+                        PMIX_RELEASE(cb);
+                        return;
+                    }
+                }
+            }
 #endif /* PMIX_ENABLE_DSTORE */
-                pmix_output_verbose(2, pmix_globals.debug_output,
-                                    "pmix_get[%d]: value retrieved from dstore", __LINE__);
-                if (PMIX_SUCCESS != (rc = process_val(val, &nvals, &results))) {
-                    cb->value_cbfunc(rc, NULL, cb->cbdata);
-                    /* cleanup */
-                    if (NULL != val) {
-                        PMIX_VALUE_RELEASE(val);
+            if (PMIX_SUCCESS != rc) {
+                /* if the user was asking about themselves, or we aren't using the dstore,
+                 * then we need to check the hash table */
+                if (PMIX_SUCCESS == (rc = pmix_hash_fetch(&nptr->modex, cb->rank, NULL, &val))) {
+                    pmix_output_verbose(2, pmix_globals.debug_output,
+                                        "pmix_get[%d]: value retrieved from hash", __LINE__);
+                    if (PMIX_SUCCESS != (rc = process_val(val, &nvals, &results))) {
+                        cb->value_cbfunc(rc, NULL, cb->cbdata);
+                        /* cleanup */
+                        if (NULL != val) {
+                            PMIX_VALUE_RELEASE(val);
+                        }
+                        PMIX_RELEASE(cb);
+                        return;
                     }
-                    PMIX_RELEASE(cb);
-                    return;
+                    PMIX_VALUE_RELEASE(val);
                 }
-                /* cleanup */
-                PMIX_VALUE_RELEASE(val);
-            } else {
+            }
+            if (PMIX_SUCCESS != rc) {
                 /* if we didn't find a modex for this rank, then we need
                  * to go get it. Thus, the caller wants -all- information for
                  * the specified rank, not just the job-level info. */
@@ -572,12 +592,17 @@ static void _getnbfn(int fd, short flags, void *cbdata)
                 PMIX_RELEASE(cb);
                 return;
             }
-            /* cleanup */
             PMIX_VALUE_RELEASE(val);
         }
         /* now let's package up the results */
         PMIX_VALUE_CREATE(val, 1);
         val->type = PMIX_DATA_ARRAY;
+        val->data.darray = (pmix_data_array_t*)malloc(sizeof(pmix_data_array_t));
+        if (NULL == val->data.darray) {
+            PMIX_VALUE_RELEASE(val);
+            cb->value_cbfunc(PMIX_ERR_NOMEM, NULL, cb->cbdata);
+            return;
+        }
         val->data.darray->type = PMIX_INFO;
         val->data.darray->size = nvals;
         PMIX_INFO_CREATE(iptr, nvals);
@@ -597,14 +622,13 @@ static void _getnbfn(int fd, short flags, void *cbdata)
                 } else {
                     pmix_value_xfer(&iptr[n].value, &info->value);
                 }
-                PMIX_INFO_FREE(info, 1);
+                PMIX_INFO_DESTRUCT(info);
             }
         }
         /* done with results array */
         PMIX_DESTRUCT(&results);
-        /* return the result to the caller */
+        /* return the result to the caller - they are responsible for releasing it */
         cb->value_cbfunc(PMIX_SUCCESS, val, cb->cbdata);
-        PMIX_VALUE_FREE(val, 1);
         PMIX_RELEASE(cb);
         return;
     }

opal/mca/pmix/pmix2x/pmix/src/util/hash.c

@@ -106,6 +106,9 @@ pmix_status_t pmix_hash_fetch(pmix_hash_table_t *table, pmix_rank_t rank,
     pmix_kval_t *hv;
     uint64_t id;
     char *node;
+    pmix_info_t *info;
+    size_t ninfo, n;
+    pmix_value_t *val;
 
     pmix_output_verbose(10, pmix_globals.debug_output,
                         "HASH:FETCH rank %d key %s",
@@ -143,7 +146,36 @@ pmix_status_t pmix_hash_fetch(pmix_hash_table_t *table, pmix_rank_t rank,
         if (NULL == key) {
             /* we will return the data as an array of pmix_info_t
              * in the kvs pmix_value_t */
-
+            val = (pmix_value_t*)malloc(sizeof(pmix_value_t));
+            if (NULL == val) {
+                return PMIX_ERR_NOMEM;
+            }
+            val->type = PMIX_DATA_ARRAY;
+            val->data.darray = (pmix_data_array_t*)malloc(sizeof(pmix_data_array_t));
+            if (NULL == val->data.darray) {
+                PMIX_VALUE_RELEASE(val);
+                return PMIX_ERR_NOMEM;
+            }
+            val->data.darray->type = PMIX_INFO;
+            val->data.darray->size = 0;
+            val->data.darray->array = NULL;
+            ninfo = pmix_list_get_size(&proc_data->data);
+            PMIX_INFO_CREATE(info, ninfo);
+            if (NULL == info) {
+                PMIX_VALUE_RELEASE(val);
+                return PMIX_ERR_NOMEM;
+            }
+            /* copy the list elements */
+            n=0;
+            PMIX_LIST_FOREACH(hv, &proc_data->data, pmix_kval_t) {
+                (void)strncpy(info[n].key, hv->key, PMIX_MAX_KEYLEN);
+                pmix_value_xfer(&info[n].value, hv->value);
+                ++n;
+            }
+            val->data.darray->size = ninfo;
+            val->data.darray->array = info;
+            *kvs = val;
+            return PMIX_SUCCESS;
         } else {
             /* find the value from within this proc_data object */
             hv = lookup_keyval(&proc_data->data, key);

opal/mca/pmix/pmix2x/pmix/test/simple/simpclient.c

@@ -269,21 +269,51 @@ int main(int argc, char **argv)
                 PMIX_VALUE_RELEASE(val);
                 free(tmp);
 
-                (void)asprintf(&tmp, "%s-%d-remote-%d", proc.nspace, n, j);
-                if (PMIX_SUCCESS != (rc = PMIx_Get(&proc, tmp, NULL, 0, &val))) {
-                    /* this data should _not_ be found as we are on the same node
-                     * and the data was "put" with a PMIX_REMOTE scope */
-                    pmix_output(0, "Client ns %s rank %d cnt %d: PMIx_Get %s returned correct", myproc.nspace, myproc.rank, j, tmp);
-                    continue;
+                if (n != myproc.rank) {
+                    (void)asprintf(&tmp, "%s-%d-remote-%d", proc.nspace, n, j);
+                    if (PMIX_SUCCESS != (rc = PMIx_Get(&proc, tmp, NULL, 0, &val))) {
+                        /* this data should _not_ be found as we are on the same node
+                         * and the data was "put" with a PMIX_REMOTE scope */
+                        pmix_output(0, "Client ns %s rank %d cnt %d: PMIx_Get %s returned correct", myproc.nspace, myproc.rank, j, tmp);
+                        continue;
+                    }
+                    pmix_output(0, "Client ns %s rank %d cnt %d: PMIx_Get %s returned remote data for a local proc",
+                                myproc.nspace, myproc.rank, j, tmp);
+                    PMIX_VALUE_RELEASE(val);
+                    free(tmp);
                 }
-                pmix_output(0, "Client ns %s rank %d cnt %d: PMIx_Get %s returned remote data for a local proc",
-                            myproc.nspace, myproc.rank, j, tmp);
-                PMIX_VALUE_RELEASE(val);
-                free(tmp);
             }
         }
     }
 
+    /* now get the data blob for myself */
+    pmix_output(0, "Client ns %s rank %d testing internal modex blob",
+                myproc.nspace, myproc.rank);
+    if (PMIX_SUCCESS == (rc = PMIx_Get(&myproc, NULL, NULL, 0, &val))) {
+        if (PMIX_DATA_ARRAY != val->type) {
+            pmix_output(0, "Client ns %s rank %d did not return an array for its internal modex blob",
+                        myproc.nspace, myproc.rank);
+            PMIX_VALUE_RELEASE(val);
+        } else if (PMIX_INFO != val->data.darray->type) {
+            pmix_output(0, "Client ns %s rank %d returned an internal modex array of type %s instead of PMIX_INFO",
+                        myproc.nspace, myproc.rank, PMIx_Data_type_string(val->data.darray->type));
+            PMIX_VALUE_RELEASE(val);
+        } else if (0 == val->data.darray->size) {
+            pmix_output(0, "Client ns %s rank %d returned an internal modex array of zero length",
+                        myproc.nspace, myproc.rank);
+            PMIX_VALUE_RELEASE(val);
+        } else {
+            pmix_info_t *iptr = (pmix_info_t*)val->data.darray->array;
+            for (n=0; n < val->data.darray->size; n++) {
+                pmix_output(0, "\tKey: %s", iptr[n].key);
+            }
+            PMIX_VALUE_RELEASE(val);
+        }
+    } else {
+        pmix_output(0, "Client ns %s rank %d internal modex blob FAILED with error %s(%d)",
+                    myproc.nspace, myproc.rank, PMIx_Error_string(rc), rc);
+    }
+
     /* log something */
     PMIX_INFO_CONSTRUCT(&info);
     (void)strncpy(info.key, "foobar", PMIX_MAX_KEYLEN);

opal/mca/pmix/pmix2x/pmix/test/test_common.c

@@ -226,10 +226,7 @@ void parse_cmd(int argc, char **argv, test_params *params)
     }
 
     // Fix rank if running under SLURM
-#if 0
-    /* the following "if" statement can never be true as rank is
-     * an unsigned 32-bit int */
-    if( 0 > params->rank ){
+    if( PMIX_RANK_UNDEF == params->rank ){
         char *ranklist = getenv("SLURM_GTIDS");
         char *rankno = getenv("SLURM_LOCALID");
         if( NULL != ranklist && NULL != rankno ){
@@ -246,7 +243,6 @@ void parse_cmd(int argc, char **argv, test_params *params)
             pmix_argv_free(argv);
         }
     }
-#endif
 
     // Fix namespace if running under SLURM
     if( NULL == params->nspace ){

opal/mca/pmix/pmix2x/pmix2x_server_south.c

@@ -112,7 +112,7 @@ int pmix2x_server_init(opal_pmix_server_module_t *module,
 
     /* convert the list to an array of pmix_info_t */
     if (NULL != info) {
-        sz = opal_list_get_size(info);
+        sz = opal_list_get_size(info) + 2;
         PMIX_INFO_CREATE(pinfo, sz);
         n = 0;
         OPAL_LIST_FOREACH(kv, info, opal_value_t) {
@@ -121,8 +121,8 @@ int pmix2x_server_init(opal_pmix_server_module_t *module,
             ++n;
         }
     } else {
-        sz = 0;
-        pinfo = NULL;
+        sz = 2;
+        PMIX_INFO_CREATE(pinfo, 2);
     }
 
     /* insert ourselves into our list of jobids - it will be the
@@ -133,6 +133,9 @@ int pmix2x_server_init(opal_pmix_server_module_t *module,
     opal_list_append(&mca_pmix_pmix2x_component.jobids, &job->super);
     OPAL_PMIX_RELEASE_THREAD(&opal_pmix_base.lock);
 
+    /* add our nspace and rank to the array going down to the PMIx server */
+    PMIX_INFO_LOAD(&pinfo[sz-2], PMIX_SERVER_NSPACE, job->nspace, PMIX_STRING);
+    PMIX_INFO_LOAD(&pinfo[sz-1], PMIX_SERVER_RANK, &OPAL_PROC_MY_NAME.vpid, PMIX_PROC_RANK);
     if (PMIX_SUCCESS != (rc = PMIx_server_init(&mymodule, pinfo, sz))) {
         PMIX_INFO_FREE(pinfo, sz);
         return pmix2x_convert_rc(rc);

opal/mca/pmix/pmix_types.h

@@ -62,6 +62,8 @@ BEGIN_C_DECLS
 #define OPAL_PMIX_CONNECT_SYSTEM_FIRST          "pmix.cnct.sys.first"   // (bool) Preferentially look for a system-level PMIx server first
 #define OPAL_PMIX_REGISTER_NODATA               "pmix.reg.nodata"       // (bool) Registration is for nspace only, do not copy job data
 #define OPAL_PMIX_SERVER_ENABLE_MONITORING      "pmix.srv.monitor"      // (bool) Enable PMIx internal monitoring by server
+#define OPAL_PMIX_SERVER_NSPACE                 "pmix.srv.nspace"       // (char*) Name of the nspace to use for this server
+#define OPAL_PMIX_SERVER_RANK                   "pmix.srv.rank"         // (uint32_t) Rank of this server
 
 
 /* identification attributes */

orte/mca/ess/base/ess_base_std_orted.c

@@ -357,7 +357,9 @@ int orte_ess_base_orted_setup(void)
     }
     /* set the event base */
     opal_pmix_base_set_evbase(orte_event_base);
-    /* setup the PMIx server */
+    /* setup the PMIx server - we need this here in case the
+     * communications infrastructure wants to register
+     * information */
     if (ORTE_SUCCESS != (ret = pmix_server_init())) {
         /* the server code already barked, so let's be quiet */
         ret = ORTE_ERR_SILENT;
@@ -398,6 +400,9 @@ int orte_ess_base_orted_setup(void)
         goto error;
     }
 
+    /* it is now safe to start the pmix server */
+    pmix_server_start();
+
     if (NULL != orte_process_info.my_hnp_uri) {
         /* extract the HNP's name so we can update the routing table */
         if (ORTE_SUCCESS != (ret = orte_rml_base_parse_uris(orte_process_info.my_hnp_uri,
@@ -444,7 +449,7 @@ int orte_ess_base_orted_setup(void)
      /* add our contact info to our proc object */
      proc->rml_uri = orte_rml.get_contact_info();
 
-   /*
+    /*
      * Group communications
      */
     if (ORTE_SUCCESS != (ret = mca_base_framework_open(&orte_grpcomm_base_framework, 0))) {