Bring the ofi/rml component online by completing the wireup protocol for the daemons. Cleanup the current confusion over how connection info gets created and

passed to make it all flow thru the opal/pmix "put/get" operations. Update the PMIx code to latest master to pickup some required behaviors.

Remove the no-longer-required get_contact_info and set_contact_info from the RML layer.

Add an MCA param to allow the ofi/rml component to route messages if desired. This is mainly for experimentation at this point as we aren't sure if routing wi
ll be beneficial at large scales. Leave it "off" by default.

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

Author: Ralph Castain

opal/mca/pmix/pmix2x/pmix2x_client.c

@@ -243,6 +243,7 @@ int pmix2x_store_local(const opal_process_name_t *proc, opal_value_t *val)
     pmix_status_t rc;
     pmix_proc_t p;
     char *nsptr;
+    opal_pmix2x_jobid_trkr_t *job;
 
     OPAL_PMIX_ACQUIRE_THREAD(&opal_pmix_base.lock);
 
@@ -254,7 +255,13 @@ int pmix2x_store_local(const opal_process_name_t *proc, opal_value_t *val)
 
     if (NULL != proc) {
         if (NULL == (nsptr = pmix2x_convert_jobid(proc->jobid))) {
-            return OPAL_ERR_NOT_FOUND;
+            job = OBJ_NEW(opal_pmix2x_jobid_trkr_t);
+            (void)opal_snprintf_jobid(job->nspace, PMIX_MAX_NSLEN, proc->jobid);
+            job->jobid = proc->jobid;
+            OPAL_PMIX_ACQUIRE_THREAD(&opal_pmix_base.lock);
+            opal_list_append(&mca_pmix_pmix2x_component.jobids, &job->super);
+            OPAL_PMIX_RELEASE_THREAD(&opal_pmix_base.lock);
+            nsptr = job->nspace;
         }
         (void)strncpy(p.nspace, nsptr, PMIX_MAX_NSLEN);
         p.rank = pmix2x_convert_opalrank(proc->vpid);

orte/mca/ess/base/ess_base_std_app.c

@@ -198,18 +198,34 @@ int orte_ess_base_app_setup(bool db_restrict_local)
         }
     }
     if (NULL != orte_process_info.my_daemon_uri) {
+        opal_value_t val;
+
         /* extract the daemon's name so we can update the routing table */
         if (ORTE_SUCCESS != (ret = orte_rml_base_parse_uris(orte_process_info.my_daemon_uri,
                                                             ORTE_PROC_MY_DAEMON, NULL))) {
             ORTE_ERROR_LOG(ret);
             error = "orte_rml_parse_daemon";
             goto error;
         }
-        /* Set the contact info in the RML - this won't actually establish
-         * the connection, but just tells the RML how to reach the daemon
+        /* Set the contact info in the database - this won't actually establish
+         * the connection, but just tells us how to reach the daemon
          * if/when we attempt to send to it
          */
-        orte_rml.set_contact_info(orte_process_info.my_daemon_uri);
+        OBJ_CONSTRUCT(&val, opal_value_t);
+        val.key = OPAL_PMIX_PROC_URI;
+        val.type = OPAL_STRING;
+        val.data.string = orte_process_info.my_daemon_uri;
+        if (OPAL_SUCCESS != (ret = opal_pmix.store_local(ORTE_PROC_MY_DAEMON, &val))) {
+            ORTE_ERROR_LOG(ret);
+            val.key = NULL;
+            val.data.string = NULL;
+            OBJ_DESTRUCT(&val);
+            error = "store DAEMON URI";
+            goto error;
+        }
+        val.key = NULL;
+        val.data.string = NULL;
+        OBJ_DESTRUCT(&val);
     }
 
     /* setup the errmgr */

orte/mca/ess/base/ess_base_std_orted.c

@@ -419,6 +419,8 @@ int orte_ess_base_orted_setup(void)
     pmix_server_start();
 
     if (NULL != orte_process_info.my_hnp_uri) {
+        opal_value_t val;
+
         /* 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,
                                                             ORTE_PROC_MY_HNP, NULL))) {
@@ -430,7 +432,21 @@ int orte_ess_base_orted_setup(void)
          * the connection, but just tells the RML how to reach the HNP
          * if/when we attempt to send to it
          */
-        orte_rml.set_contact_info(orte_process_info.my_hnp_uri);
+        OBJ_CONSTRUCT(&val, opal_value_t);
+        val.key = OPAL_PMIX_PROC_URI;
+        val.type = OPAL_STRING;
+        val.data.string = orte_process_info.my_hnp_uri;
+        if (OPAL_SUCCESS != (ret = opal_pmix.store_local(ORTE_PROC_MY_HNP, &val))) {
+            ORTE_ERROR_LOG(ret);
+            val.key = NULL;
+            val.data.string = NULL;
+            OBJ_DESTRUCT(&val);
+            error = "store HNP URI";
+            goto error;
+        }
+        val.key = NULL;
+        val.data.string = NULL;
+        OBJ_DESTRUCT(&val);
     }
 
     /* select the errmgr */
@@ -461,9 +477,6 @@ int orte_ess_base_orted_setup(void)
     }
     OPAL_LIST_DESTRUCT(&transports);
 
-     /* add our contact info to our proc object */
-     proc->rml_uri = orte_rml.get_contact_info();
-
     /*
      * Group communications
      */
@@ -539,7 +552,7 @@ int orte_ess_base_orted_setup(void)
         }
     }
 
-    if (orte_static_ports) {
+    if (orte_static_ports || orte_fwd_mpirun_port) {
         if (NULL == orte_node_regex) {
             /* we didn't get the node info */
             error = "cannot construct daemon map for static ports - no node map info";

orte/mca/ess/hnp/ess_hnp_module.c

@@ -471,7 +471,10 @@ static int rte_init(void)
     proc->name.jobid = ORTE_PROC_MY_NAME->jobid;
     proc->name.vpid = ORTE_PROC_MY_NAME->vpid;
     proc->pid = orte_process_info.pid;
-    proc->rml_uri = orte_rml.get_contact_info();
+    orte_oob_base_get_addr(&proc->rml_uri);
+    orte_process_info.my_hnp_uri = strdup(proc->rml_uri);
+    /* we are also officially a daemon, so better update that field too */
+    orte_process_info.my_daemon_uri = strdup(proc->rml_uri);
     proc->state = ORTE_PROC_STATE_RUNNING;
     OBJ_RETAIN(node);  /* keep accounting straight */
     proc->node = node;
@@ -615,15 +618,6 @@ static int rte_init(void)
         goto error;
     }
 
-    /* we are an hnp, so update the contact info field for later use */
-    orte_process_info.my_hnp_uri = orte_rml.get_contact_info();
-    if (NULL != proc->rml_uri) {
-        free(proc->rml_uri);
-    }
-    proc->rml_uri = strdup(orte_process_info.my_hnp_uri);
-
-    /* we are also officially a daemon, so better update that field too */
-    orte_process_info.my_daemon_uri = strdup(orte_process_info.my_hnp_uri);
     /* setup the orte_show_help system to recv remote output */
     orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, ORTE_RML_TAG_SHOW_HELP,
                             ORTE_RML_PERSISTENT, orte_show_help_recv, NULL);

orte/mca/grpcomm/direct/grpcomm_direct.c

@@ -23,6 +23,7 @@
 
 #include "opal/dss/dss.h"
 #include "opal/class/opal_list.h"
+#include "opal/mca/pmix/pmix.h"
 
 #include "orte/mca/errmgr/errmgr.h"
 #include "orte/mca/rml/base/base.h"
@@ -273,6 +274,9 @@ static void xcast_recv(int status, orte_process_name_t* sender,
     char *rtmod, *nidmap;
     size_t inlen, cmplen;
     uint8_t *packed_data, *cmpdata;
+    int32_t nvals, i;
+    opal_value_t *kv;
+    orte_process_name_t dmn;
 
     OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_framework.framework_output,
                          "%s grpcomm:direct:xcast:recv: with %d bytes",
@@ -446,35 +450,49 @@ static void xcast_recv(int status, orte_process_name_t* sender,
                     /* routing is now possible */
                     orte_routed_base.routing_enabled = true;
 
-                    /* see if we have wiring info as well */
+                    /* unpack the byte object */
                     cnt=1;
-                    if (ORTE_SUCCESS != (ret = opal_dss.unpack(data, &flag, &cnt, OPAL_INT8))) {
+                    if (ORTE_SUCCESS != (ret = opal_dss.unpack(data, &bo, &cnt, OPAL_BYTE_OBJECT))) {
                         ORTE_ERROR_LOG(ret);
                         goto relay;
                     }
-
-                    if (0 != flag) {
-                        /* unpack the byte object */
+                    if (0 < bo->size) {
+                        /* load it into a buffer */
+                        OBJ_CONSTRUCT(&wireup, opal_buffer_t);
+                        opal_dss.load(&wireup, bo->bytes, bo->size);
+                        /* decode it, pushing the info into our database */
                         cnt=1;
-                        if (ORTE_SUCCESS != (ret = opal_dss.unpack(data, &bo, &cnt, OPAL_BYTE_OBJECT))) {
-                            ORTE_ERROR_LOG(ret);
-                            goto relay;
-                        }
-                        if (0 < bo->size) {
-                            /* load it into a buffer */
-                            OBJ_CONSTRUCT(&wireup, opal_buffer_t);
-                            opal_dss.load(&wireup, bo->bytes, bo->size);
-                            /* pass it for processing */
-                            if (ORTE_SUCCESS != (ret = orte_rml_base_update_contact_info(&wireup))) {
+                        while (OPAL_SUCCESS == (ret = opal_dss.unpack(&wireup, &dmn, &cnt, ORTE_NAME))) {
+                            cnt = 1;
+                            if (ORTE_SUCCESS != (ret = opal_dss.unpack(&wireup, &nvals, &cnt, OPAL_INT32))) {
                                 ORTE_ERROR_LOG(ret);
-                                OBJ_DESTRUCT(&wireup);
-                                goto relay;
+                                break;
                             }
-                            /* done with the wireup buffer - dump it */
-                            OBJ_DESTRUCT(&wireup);
+                            for (i=0; i < nvals; i++) {
+                                cnt = 1;
+                                if (ORTE_SUCCESS != (ret = opal_dss.unpack(&wireup, &kv, &cnt, OPAL_VALUE))) {
+                                    ORTE_ERROR_LOG(ret);
+                                    break;
+                                }
+                                OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_framework.framework_output,
+                                                    "%s STORING MODEX DATA FOR PROC %s KEY %s",
+                                                    ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
+                                                    ORTE_NAME_PRINT(&dmn), kv->key));
+                                if (OPAL_SUCCESS != (ret = opal_pmix.store_local(&dmn, kv))) {
+                                    ORTE_ERROR_LOG(ret);
+                                    OBJ_RELEASE(kv);
+                                    break;
+                                }
+                                OBJ_RELEASE(kv);
+                            }
+                        }
+                        if (ORTE_ERR_UNPACK_READ_PAST_END_OF_BUFFER != ret) {
+                            ORTE_ERROR_LOG(ret);
                         }
-                        free(bo);
+                        /* done with the wireup buffer - dump it */
+                        OBJ_DESTRUCT(&wireup);
                     }
+                    free(bo);
                 }
                 /* copy the remainder of the payload - we don't pass wiring info
                  * to the odls */
@@ -499,7 +517,7 @@ static void xcast_recv(int status, orte_process_name_t* sender,
         opal_dss.copy_payload(relay, data);
     }
 
- relay:
+  relay:
     if (!orte_do_not_launch) {
         /* get the list of next recipients from the routed module */
         orte_routed.get_routing_list(rtmod, &coll);

orte/mca/odls/base/odls_base_default_fns.c

@@ -106,7 +106,7 @@
 int orte_odls_base_default_get_add_procs_data(opal_buffer_t *buffer,
                                               orte_jobid_t job)
 {
-    int rc;
+    int rc, v;
     orte_job_t *jdata=NULL, *jptr;
     orte_job_map_t *map=NULL;
     opal_buffer_t *wireup, jobdata;
@@ -116,6 +116,9 @@ int orte_odls_base_default_get_add_procs_data(opal_buffer_t *buffer,
     void *nptr;
     uint32_t key;
     char *nidmap;
+    orte_proc_t *dmn;
+    opal_value_t *val = NULL, *kv;
+    opal_list_t *modex;
 
     /* get the job data pointer */
     if (NULL == (jdata = orte_get_job_data_object(job))) {
@@ -156,36 +159,106 @@ int orte_odls_base_default_get_add_procs_data(opal_buffer_t *buffer,
             ORTE_ERROR_LOG(rc);
             return rc;
         }
-        if (!orte_static_ports && !orte_fwd_mpirun_port) {
-            /* pack a flag indicating wiring info is provided */
-            flag = 1;
-            opal_dss.pack(buffer, &flag, 1, OPAL_INT8);
-            /* get wireup info for daemons per the selected routing module */
-            wireup = OBJ_NEW(opal_buffer_t);
-            if (ORTE_SUCCESS != (rc = orte_rml_base_get_contact_info(ORTE_PROC_MY_NAME->jobid, wireup))) {
+        /* get wireup info for daemons */
+        if (NULL == (jptr = orte_get_job_data_object(ORTE_PROC_MY_NAME->jobid))) {
+            ORTE_ERROR_LOG(ORTE_ERR_BAD_PARAM);
+            return ORTE_ERR_BAD_PARAM;
+        }
+        wireup = OBJ_NEW(opal_buffer_t);
+        /* always include data for mpirun as the daemons can't have it yet */
+        val = NULL;
+        if (OPAL_SUCCESS != (rc = opal_pmix.get(ORTE_PROC_MY_NAME, NULL, NULL, &val)) || NULL == val) {
+            ORTE_ERROR_LOG(rc);
+            OBJ_RELEASE(wireup);
+            return rc;
+        } else {
+            /* the data is returned as a list of key-value pairs in the opal_value_t */
+            if (OPAL_PTR != val->type) {
+                ORTE_ERROR_LOG(ORTE_ERR_NOT_FOUND);
+                OBJ_RELEASE(wireup);
+                return ORTE_ERR_NOT_FOUND;
+            }
+            if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, ORTE_PROC_MY_NAME, 1, ORTE_NAME))) {
                 ORTE_ERROR_LOG(rc);
                 OBJ_RELEASE(wireup);
                 return rc;
             }
-            /* put it in a byte object for xmission */
-            opal_dss.unload(wireup, (void**)&bo.bytes, &numbytes);
-            /* pack the byte object - zero-byte objects are fine */
-            bo.size = numbytes;
-            boptr = &bo;
-            if (ORTE_SUCCESS != (rc = opal_dss.pack(buffer, &boptr, 1, OPAL_BYTE_OBJECT))) {
+            modex = (opal_list_t*)val->data.ptr;
+            numbytes = (int32_t)opal_list_get_size(modex);
+            if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, &numbytes, 1, OPAL_INT32))) {
                 ORTE_ERROR_LOG(rc);
                 OBJ_RELEASE(wireup);
                 return rc;
             }
-            /* release the data since it has now been copied into our buffer */
-            if (NULL != bo.bytes) {
-                free(bo.bytes);
+            OPAL_LIST_FOREACH(kv, modex, opal_value_t) {
+                if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, &kv, 1, OPAL_VALUE))) {
+                    ORTE_ERROR_LOG(rc);
+                    OBJ_RELEASE(wireup);
+                    return rc;
+                }
             }
-            OBJ_RELEASE(wireup);
-        } else {
-            /* pack a flag indicating no wireup data is provided */
-            flag = 0;
-            opal_dss.pack(buffer, &flag, 1, OPAL_INT8);
+            OPAL_LIST_RELEASE(modex);
+            OBJ_RELEASE(val);
+        }
+        /* if we didn't rollup the connection info, then we have
+         * to provide a complete map of connection info */
+        if (!orte_static_ports && !orte_fwd_mpirun_port) {
+            for (v=1; v < jptr->procs->size; v++) {
+                if (NULL == (dmn = (orte_proc_t*)opal_pointer_array_get_item(jptr->procs, v))) {
+                    continue;
+                }
+                val = NULL;
+                if (OPAL_SUCCESS != (rc = opal_pmix.get(&dmn->name, NULL, NULL, &val)) || NULL == val) {
+                    ORTE_ERROR_LOG(rc);
+                    OBJ_RELEASE(buffer);
+                    return rc;
+                } else {
+                    /* the data is returned as a list of key-value pairs in the opal_value_t */
+                    if (OPAL_PTR != val->type) {
+                        ORTE_ERROR_LOG(ORTE_ERR_NOT_FOUND);
+                        OBJ_RELEASE(buffer);
+                        return ORTE_ERR_NOT_FOUND;
+                    }
+                    if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, &dmn->name, 1, ORTE_NAME))) {
+                        ORTE_ERROR_LOG(rc);
+                        OBJ_RELEASE(buffer);
+                        OBJ_RELEASE(wireup);
+                        return rc;
+                    }
+                    modex = (opal_list_t*)val->data.ptr;
+                    numbytes = (int32_t)opal_list_get_size(modex);
+                    if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, &numbytes, 1, OPAL_INT32))) {
+                        ORTE_ERROR_LOG(rc);
+                        OBJ_RELEASE(buffer);
+                        OBJ_RELEASE(wireup);
+                        return rc;
+                    }
+                    OPAL_LIST_FOREACH(kv, modex, opal_value_t) {
+                        if (ORTE_SUCCESS != (rc = opal_dss.pack(wireup, &kv, 1, OPAL_VALUE))) {
+                            ORTE_ERROR_LOG(rc);
+                            OBJ_RELEASE(buffer);
+                            OBJ_RELEASE(wireup);
+                            return rc;
+                        }
+                    }
+                    OPAL_LIST_RELEASE(modex);
+                    OBJ_RELEASE(val);
+                }
+            }
+        }
+        /* put it in a byte object for xmission */
+        opal_dss.unload(wireup, (void**)&bo.bytes, &numbytes);
+        OBJ_RELEASE(wireup);
+        /* pack the byte object - zero-byte objects are fine */
+        bo.size = numbytes;
+        boptr = &bo;
+        if (ORTE_SUCCESS != (rc = opal_dss.pack(buffer, &boptr, 1, OPAL_BYTE_OBJECT))) {
+            ORTE_ERROR_LOG(rc);
+            return rc;
+        }
+        /* release the data since it has now been copied into our buffer */
+        if (NULL != bo.bytes) {
+            free(bo.bytes);
         }
 
         /* we need to ensure that any new daemons get a complete