Merge pull request #4376 from rhc54/topic/interlib

Update the interlib example to show an alternative method for model declaration

Author: Ralph Castain

ompi/interlib/interlib.c

@@ -62,19 +62,18 @@ static void model_callback(int status,
 {
     opal_value_t *val;
 
-    /* we can ignore our own callback as we obviously
-     * know that we are MPI */
-    if (NULL != info) {
-        OPAL_LIST_FOREACH(val, info, opal_value_t) {
-            if (OPAL_STRING == val->type) {
-#if 0
-                opal_output(0, "OMPI Model Callback Key: %s Val %s", val->key, val->data.string);
-#else
-                if (0 == strcmp(val->key, OPAL_PMIX_MODEL_LIBRARY_NAME) &&
-                    0 == strcmp(val->data.string, "OpenMPI")) {
+    if (NULL != getenv("OMPI_SHOW_MODEL_CALLBACK")) {
+        /* we can ignore our own callback as we obviously
+         * know that we are MPI */
+        if (NULL != info) {
+            OPAL_LIST_FOREACH(val, info, opal_value_t) {
+                if (0 == strcmp(val->key, OPAL_PMIX_PROGRAMMING_MODEL) &&
+                    0 == strcmp(val->data.string, "MPI")) {
                     goto cback;
                 }
-#endif
+                if (OPAL_STRING == val->type) {
+                        opal_output(0, "OMPI Model Callback Key: %s Val %s", val->key, val->data.string);
+                }
             }
         }
     }

opal/mca/pmix/pmix3x/pmix/src/common/pmix_strings.c

@@ -139,6 +139,8 @@ PMIX_EXPORT const char* PMIx_Data_range_string(pmix_data_range_t range)
             return "AVAIL TO ANYONE WITH AUTHORIZATION";
         case PMIX_RANGE_CUSTOM:
             return "AVAIL AS SPECIFIED IN DIRECTIVES";
+        case PMIX_RANGE_PROC_LOCAL:
+            return "AVAIL ON LOCAL PROC ONLY";
         default:
             return "UNKNOWN";
     }

opal/mca/pmix/pmix3x/pmix/src/event/pmix_event_notification.c

@@ -117,9 +117,9 @@ static pmix_status_t notify_server_of_event(pmix_status_t status,
     pmix_notify_caddy_t *cd, *rbout;
 
     pmix_output_verbose(2, pmix_globals.debug_output,
-                        "client: notifying server %s:%d of status %s",
+                        "client: notifying server %s:%d of status %s for range %s",
                         pmix_globals.myid.nspace, pmix_globals.myid.rank,
-                        PMIx_Error_string(status));
+                        PMIx_Error_string(status), PMIx_Data_range_string(range));
 
     if (PMIX_RANGE_PROC_LOCAL != range) {
         /* create the msg object */

opal/mca/pmix/pmix3x/pmix3x.c

@@ -578,6 +578,8 @@ pmix_data_range_t pmix3x_convert_opalrange(opal_pmix_data_range_t range) {
         return PMIX_RANGE_GLOBAL;
     case OPAL_PMIX_RANGE_CUSTOM:
         return PMIX_RANGE_CUSTOM;
+    case OPAL_PMIX_RANGE_PROC_LOCAL:
+        return PMIX_RANGE_PROC_LOCAL;
     default:
         return PMIX_SCOPE_UNDEF;
     }

opal/mca/pmix/pmix_types.h

@@ -401,16 +401,16 @@ typedef enum {
 } opal_pmix_scope_t;
 
 /* define a range for data "published" by PMI */
-#define OPAL_PMIX_DATA_RANGE OPAL_UINT
-typedef enum {
-    OPAL_PMIX_RANGE_UNDEF = 0,
-    OPAL_PMIX_RANGE_RM,          // data is intended for the host resource manager
-    OPAL_PMIX_RANGE_LOCAL,       // available on local node only
-    OPAL_PMIX_RANGE_NAMESPACE,   // data is available to procs in the same nspace only
-    OPAL_PMIX_RANGE_SESSION,     // data available to all procs in session
-    OPAL_PMIX_RANGE_GLOBAL,      // data available to all procs
-    OPAL_PMIX_RANGE_CUSTOM       // range is specified in a opal_value_t
-} opal_pmix_data_range_t;
+#define OPAL_PMIX_DATA_RANGE OPAL_UINT8
+typedef uint8_t opal_pmix_data_range_t;
+#define OPAL_PMIX_RANGE_UNDEF        0
+#define OPAL_PMIX_RANGE_RM           1   // data is intended for the host resource manager
+#define OPAL_PMIX_RANGE_LOCAL        2   // available on local node only
+#define OPAL_PMIX_RANGE_NAMESPACE    3   // data is available to procs in the same nspace only
+#define OPAL_PMIX_RANGE_SESSION      4   // data available to all procs in session
+#define OPAL_PMIX_RANGE_GLOBAL       5   // data available to all procs
+#define OPAL_PMIX_RANGE_CUSTOM       6   // range is specified in a pmix_info_t
+#define OPAL_PMIX_RANGE_PROC_LOCAL   7   // restrict range to the local proc
 
 /* define a "persistence" policy for data published by clients */
 typedef enum {

orte/test/mpi/interlib.c

@@ -35,18 +35,15 @@ static void model_callback(int status,
     opal_value_t *val;
 
     /* we can ignore our own callback as we obviously
-     * know that we are MPI */
+     * know that we are OpenMP */
     if (NULL != info) {
         OPAL_LIST_FOREACH(val, info, opal_value_t) {
+            if (0 == strcmp(val->key, OPAL_PMIX_PROGRAMMING_MODEL) &&
+                0 == strcmp(val->data.string, "OpenMP")) {
+                goto cback;
+            }
             if (OPAL_STRING == val->type) {
-#if 1
                 opal_output(0, "Thread Model Callback Key: %s Val %s", val->key, val->data.string);
-#else
-                if (0 == strcmp(val->key, OPAL_PMIX_MODEL_LIBRARY_NAME) &&
-                    0 == strcmp(val->data.string, "OpenMPI")) {
-                    goto cback;
-                }
-#endif
             }
         }
     }
@@ -62,12 +59,39 @@ static void model_callback(int status,
     OPAL_PMIX_WAKEUP_THREAD(&thread_complete);
 }
 
+static void opcbfunc(int status, void *cbdata)
+{
+    opal_pmix_lock_t *lock = (opal_pmix_lock_t*)cbdata;
+    OPAL_PMIX_WAKEUP_THREAD(lock);
+}
+
+static void infocb(int status,
+                   opal_list_t *info,
+                   void *cbdata,
+                   opal_pmix_release_cbfunc_t release_fn,
+                   void *release_cbdata)
+{
+    opal_pmix_lock_t *lock = (opal_pmix_lock_t*)cbdata;
+    opal_value_t *kv;
+
+    OPAL_LIST_FOREACH(kv, info, opal_value_t) {
+        opal_output(0, "QUERY DATA KEY: %s VALUE %s", kv->key, kv->data.string);
+    }
+    if (NULL != release_fn) {
+        release_fn(release_cbdata);
+    }
+    OPAL_PMIX_WAKEUP_THREAD(lock);
+}
+
 static void *mylib(void *ptr)
 {
     opal_list_t info, directives;
     opal_value_t *kv;
     int ret;
     opal_pmix_lock_t lock;
+    bool init = false;
+    opal_pmix_query_t *query;
+    opal_pmix_pdata_t *pdata;
 
     OPAL_PMIX_CONSTRUCT_LOCK(&thread_complete);
 
@@ -94,9 +118,31 @@ static void *mylib(void *ptr)
     kv->data.string = strdup("PTHREAD");
     opal_list_append(&info, &kv->super);
 
-    /* call pmix to initialize these values */
-    ret = opal_pmix.init(&info);
-    OPAL_LIST_DESTRUCT(&info);
+    /* see if pmix is already initialized */
+    if (opal_pmix.initialized()) {
+        /* mark that this isn't to go to any default event handler - pmix_init
+         * takes care of that for us, but we have to explicitly do it here */
+        kv = OBJ_NEW(opal_value_t);
+        kv->key = strdup(OPAL_PMIX_EVENT_NON_DEFAULT);
+        kv->type = OPAL_BOOL;
+        kv->data.flag = true;
+        opal_list_append(&info, &kv->super);
+        /* it is, so let's just use the event notification
+         * API to let everyone know we are here */
+        OPAL_PMIX_CONSTRUCT_LOCK(&lock);
+        ret = opal_pmix.notify_event(OPAL_ERR_MODEL_DECLARED,
+                                     &orte_process_info.my_name,
+                                     OPAL_PMIX_RANGE_PROC_LOCAL, &info,
+                                     opcbfunc, &lock);
+        OPAL_PMIX_WAIT_THREAD(&lock);
+        OPAL_PMIX_DESTRUCT_LOCK(&lock);
+        OPAL_LIST_DESTRUCT(&info);
+    } else {
+        /* call pmix to initialize these values */
+        ret = opal_pmix.init(&info);
+        OPAL_LIST_DESTRUCT(&info);
+        init = true;
+    }
 
     /* register to receive model callbacks */
 
@@ -124,12 +170,61 @@ static void *mylib(void *ptr)
                                  (void*)&lock);
     OPAL_PMIX_WAIT_THREAD(&lock);
     OPAL_PMIX_DESTRUCT_LOCK(&lock);
+    OPAL_LIST_DESTRUCT(&info);
+    OPAL_LIST_DESTRUCT(&directives);
 
     /* wait for the model callback */
     OPAL_PMIX_WAIT_THREAD(&thread_complete);
 
-    /* finalize */
-    opal_pmix.finalize();
+    /* let's do a couple of operations just to verify we can,
+     * starting with a query */
+    OBJ_CONSTRUCT(&info, opal_list_t);
+    query = OBJ_NEW(opal_pmix_query_t);
+    opal_argv_append_nosize(&query->keys, OPAL_PMIX_QUERY_NAMESPACES);
+    opal_list_append(&info, &query->super);
+    OPAL_PMIX_CONSTRUCT_LOCK(&lock);
+    opal_pmix.query(&info, infocb, &lock);
+    OPAL_PMIX_WAIT_THREAD(&lock);
+    OPAL_PMIX_DESTRUCT_LOCK(&lock);
+    OPAL_LIST_DESTRUCT(&info);
+
+    /* Get something */
+    opal_pmix.get(&orte_process_info.my_name,
+                  "WASSUP", NULL, &kv);
+    if (NULL == kv) {
+        fprintf(stderr, "ERROR GETTING WASSUP\n");
+    } else {
+        fprintf(stderr, "THREAD WASSUP: %s\n", kv->data.string);
+        OBJ_RELEASE(kv);
+    }
+
+    /* lookup something published by the main thread */
+    OBJ_CONSTRUCT(&info, opal_list_t);
+    pdata = OBJ_NEW(opal_pmix_pdata_t);
+    pdata->proc = orte_process_info.my_name;
+    pdata->value.key = strdup("SOMETHING");
+    opal_list_append(&info, &pdata->super);
+    /* tell the call to wait for the data to be published */
+    OBJ_CONSTRUCT(&directives, opal_list_t);
+    kv = OBJ_NEW(opal_value_t);
+    kv->key = strdup(OPAL_PMIX_WAIT);
+    kv->type = OPAL_INT;
+    kv->data.integer = 0;  // wait for all
+    opal_list_append(&directives, &kv->super);
+
+    if (OPAL_SUCCESS != opal_pmix.lookup(&info, &directives)) {
+        fprintf(stderr, "LOOKUP FAILED\n");
+    } else {
+        pdata = (opal_pmix_pdata_t*)opal_list_get_first(&info);
+        fprintf(stderr, "LOOKUP RETURNED %s\n", pdata->value.data.string);
+    }
+    OPAL_LIST_DESTRUCT(&info);
+    OPAL_LIST_DESTRUCT(&directives);
+
+    if (init) {
+        /* need to finalize to maintain refcount */
+        opal_pmix.finalize();
+    }
 
     /* done */
     return NULL;
@@ -142,12 +237,23 @@ int main(int argc, char* argv[])
     char *bindings = NULL;
     pid_t pid;
     pthread_t mythread;
+    opal_value_t kv, *kptr;
+    opal_list_t list;
 
     MPI_Init(&argc, &argv);
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
     pid = getpid();
 
+    /* push something the thread can recognize */
+    OBJ_CONSTRUCT(&kv, opal_value_t);
+    kv.key = strdup("WASSUP");
+    kv.type = OPAL_STRING;
+    kv.data.string = strdup("nothing");
+    opal_pmix.put(OPAL_PMIX_LOCAL, &kv);
+    OBJ_DESTRUCT(&kv);
+    /* no need to commit it as this is strictly within ourselves */
+
     /* spin up a thread */
     if (pthread_create(&mythread, NULL, mylib, NULL)) {
         fprintf(stderr, "Error creating thread\n");
@@ -166,6 +272,16 @@ int main(int argc, char* argv[])
            rank, size, orte_process_info.num_local_peers, rc,
            (NULL == bindings) ? "NULL" : bindings);
 
+    /* publish something */
+    OBJ_CONSTRUCT(&list, opal_list_t);
+    kptr = OBJ_NEW(opal_value_t);
+    kptr->key = strdup("SOMETHING");
+    kptr->type = OPAL_STRING;
+    kptr->data.string = strdup("SILLY-THING");
+    opal_list_append(&list, &kptr->super);
+    opal_pmix.publish(&list);
+    OPAL_LIST_DESTRUCT(&list);
+
     /* wait for the thread to finish */
     if (pthread_join(mythread, NULL)) {
         fprintf(stderr, "Error joining thread\n");