Merge pull request #2669 from rhc54/topic/memprobe

Complete the memprobe support.

Author: Ralph Castain

contrib/scaling/Makefile

@@ -4,14 +4,14 @@ all: $(PROGS)
 
 CFLAGS = -O
 
-orte_no_op:
+orte_no_op: orte_no_op.c
 	ortecc -o orte_no_op orte_no_op.c
 
-mpi_no_op:
+mpi_no_op: mpi_no_op.c
 	mpicc -o mpi_no_op mpi_no_op.c
 
-mpi_memprobe:
-	mpicc -o mpi_memprobe mpi_memprobe.c -lopen-pal
+mpi_memprobe: mpi_memprobe.c
+	mpicc -o mpi_memprobe mpi_memprobe.c -lopen-pal -lopen-rte
 
 clean:
 	rm -f $(PROGS) *~

contrib/scaling/mpi_memprobe.c

@@ -17,7 +17,6 @@
 #include "orte/mca/errmgr/errmgr.h"
 
 static int rank, size;
-static volatile int active;
 static volatile bool wait_for_release = true;
 #define MEMPROBE_RELEASE 12345
 
@@ -27,7 +26,6 @@ static void _release_fn(int status,
                         opal_pmix_notification_complete_fn_t cbfunc,
                         void *cbdata)
 {
-    fprintf(stderr, "Rank %d: Release recvd\n", rank);
     /* must let the notifier know we are done */
     if (NULL != cbfunc) {
         cbfunc(OPAL_ERR_HANDLERS_COMPLETE, NULL, NULL, NULL, cbdata);
@@ -58,7 +56,6 @@ static void qcbfunc(int status,
     opal_list_t *results = (opal_list_t*)cbdata;
     opal_value_t *kv;
 
-    fprintf(stderr, "Rank %d: Query returned status %d\n", rank, status);
     if (NULL != info) {
         while (NULL != (kv = (opal_value_t*)opal_list_remove_first(info))) {
             opal_list_append(results, &kv->super);
@@ -70,61 +67,90 @@ static void qcbfunc(int status,
     wait_for_release = false;
 }
 
-int main(int argc, char* argv[])
+static void notifycbfunc(int status, void *cbdata)
 {
-    opal_list_t *codes;
-    opal_value_t *kv;
-    opal_pmix_query_t *q;
-    opal_list_t query, response;
-
-    MPI_Init(&argc, &argv);
-    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
-    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    volatile int *active = (volatile int*)cbdata;
+    *active = status;
+}
 
-    /* everyone registers their event handler */
-    codes = OBJ_NEW(opal_list_t);
+static void sample(void)
+{
+    opal_value_t *kv, *ival;
+    opal_pmix_query_t *q;
+    opal_list_t query, response, *lt;
+    volatile int active;
+    char **answer = NULL, *tmp, *msg;
+
+    OBJ_CONSTRUCT(&query, opal_list_t);
+    OBJ_CONSTRUCT(&response, opal_list_t);
+    q = OBJ_NEW(opal_pmix_query_t);
+    opal_list_append(&query, &q->super);
+    opal_argv_append_nosize(&q->keys, OPAL_PMIX_QUERY_MEMORY_USAGE);
+    /* qualify that we just want local avg, min/max values reported */
     kv = OBJ_NEW(opal_value_t);
-    kv->key = strdup("errorcode");
-    kv->type = OPAL_INT;
-    kv->data.integer = MEMPROBE_RELEASE;
-    opal_list_append(codes, &kv->super);
+    kv->key = strdup(OPAL_PMIX_QUERY_LOCAL_ONLY);
+    kv->type = OPAL_BOOL;
+    kv->data.flag = true;
+    opal_list_append(&q->qualifiers, &kv->super);
+    kv = OBJ_NEW(opal_value_t);
+    kv->key = strdup(OPAL_PMIX_QUERY_REPORT_AVG);
+    kv->type = OPAL_BOOL;
+    kv->data.flag = true;
+    opal_list_append(&q->qualifiers, &kv->super);
+    kv = OBJ_NEW(opal_value_t);
+    kv->key = strdup(OPAL_PMIX_QUERY_REPORT_MINMAX);
+    kv->type = OPAL_BOOL;
+    kv->data.flag = true;
+    opal_list_append(&q->qualifiers, &kv->super);
+    /* issue the request */
+    wait_for_release = true;
+    opal_pmix.query(&query, qcbfunc, (void*)&response);
+    /* wait for the query to complete */
+    while (wait_for_release) {
+        usleep(10);
+    }
+    wait_for_release = true;
+    /* log my own results as a single string so the output
+     * doesn't get garbled on the other end */
+    asprintf(&tmp, "Data for node %s", orte_process_info.nodename);
+    opal_argv_append_nosize(&answer, tmp);
+    free(tmp);
+    OPAL_LIST_FOREACH(kv, &response, opal_value_t) {
+        lt = (opal_list_t*)kv->data.ptr;
+        OPAL_LIST_FOREACH(ival, lt, opal_value_t) {
+            if (0 == strcmp(ival->key, OPAL_PMIX_DAEMON_MEMORY)) {
+                asprintf(&tmp, "\tDaemon: %f", ival->data.fval);
+                opal_argv_append_nosize(&answer, tmp);
+                free(tmp);
+            } else if (0 == strcmp(ival->key, OPAL_PMIX_CLIENT_AVG_MEMORY)) {
+                asprintf(&tmp, "\tClient: %f", ival->data.fval);
+                opal_argv_append_nosize(&answer, tmp);
+                free(tmp);
+            } else {
+                fprintf(stderr, "\tUnknown key: %s", ival->key);
+            }
+        }
+    }
+    opal_argv_append_nosize(&answer, "\n");
+    OPAL_LIST_DESTRUCT(&response);
 
+    /* construct the log output */
+    OBJ_CONSTRUCT(&response, opal_list_t);
+    kv = OBJ_NEW(opal_value_t);
+    kv->key = strdup(OPAL_PMIX_LOG_STDOUT);
+    kv->type = OPAL_STRING;
+    kv->data.string = opal_argv_join(answer, '\n');
+    opal_list_append(&response, &kv->super);
+    opal_argv_free(answer);
     active = -1;
-    opal_pmix.register_evhandler(codes, NULL, _release_fn, _register_fn, (void*)&active);
+    opal_pmix.log(&response, notifycbfunc, (void*)&active);
     while (-1 == active) {
         usleep(10);
     }
+    OPAL_LIST_DESTRUCT(&response);
+
 
-    /* rank 0 asks for memory to be sampled, while everyone else waits */
     if (0 == rank) {
-        fprintf(stderr, "Sampling memory usage after MPI_Init\n");
-        OBJ_CONSTRUCT(&query, opal_list_t);
-        OBJ_CONSTRUCT(&response, opal_list_t);
-        q = OBJ_NEW(opal_pmix_query_t);
-        opal_list_append(&query, &q->super);
-        opal_argv_append_nosize(&q->keys, OPAL_PMIX_QUERY_MEMORY_USAGE);
-        /* qualify that we just want avg, min/max values reported */
-        kv = OBJ_NEW(opal_value_t);
-        kv->key = strdup(OPAL_PMIX_QUERY_REPORT_AVG);
-        kv->type = OPAL_BOOL;
-        kv->data.flag = true;
-        opal_list_append(&q->qualifiers, &kv->super);
-        kv = OBJ_NEW(opal_value_t);
-        kv->key = strdup(OPAL_PMIX_QUERY_REPORT_MINMAX);
-        kv->type = OPAL_BOOL;
-        kv->data.flag = true;
-        opal_list_append(&q->qualifiers, &kv->super);
-        /* issue the request */
-        wait_for_release = true;
-        opal_pmix.query(&query, qcbfunc, (void*)&response);
-        while (wait_for_release) {
-            usleep(10);
-        }
-        /* output the results */
-        OPAL_LIST_FOREACH(kv, &response, opal_value_t) {
-            fprintf(stderr, "\tResults: %s\n", kv->key);
-        }
-        OPAL_LIST_DESTRUCT(&response);
         /* send the notification to release the other procs */
         wait_for_release = true;
         OBJ_CONSTRUCT(&response, opal_list_t);
@@ -133,16 +159,58 @@ int main(int argc, char* argv[])
         kv->type = OPAL_BOOL;
         kv->data.flag = true;
         opal_list_append(&response, &kv->super);
+        active = -1;
         if (OPAL_SUCCESS != opal_pmix.notify_event(MEMPROBE_RELEASE, NULL,
                                                    OPAL_PMIX_RANGE_GLOBAL, &response,
-                                                   NULL, NULL)) {
+                                                   notifycbfunc, (void*)&active)) {
             fprintf(stderr, "Notify event failed\n");
             exit(1);
         }
-        while (wait_for_release) {
+        while (-1 == active) {
             usleep(10);
         }
         OPAL_LIST_DESTRUCT(&response);
+    }
+
+    /* now wait for notification */
+    while (wait_for_release) {
+        usleep(10);
+    }
+}
+
+int main(int argc, char* argv[])
+{
+    opal_list_t *codes;
+    opal_value_t *kv;
+    volatile int active;
+
+    MPI_Init(&argc, &argv);
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+    if (0 == rank) {
+        fprintf(stderr, "Sampling memory usage after MPI_Init\n");
+    }
+
+    /* everyone registers their event handler */
+    codes = OBJ_NEW(opal_list_t);
+    kv = OBJ_NEW(opal_value_t);
+    kv->key = strdup("errorcode");
+    kv->type = OPAL_INT;
+    kv->data.integer = MEMPROBE_RELEASE;
+    opal_list_append(codes, &kv->super);
+
+    active = -1;
+    opal_pmix.register_evhandler(codes, NULL, _release_fn, _register_fn, (void*)&active);
+    while (-1 == active) {
+        usleep(10);
+    }
+
+    /* if I am the local leader (i.e., local_rank=0), then I ask
+     * my daemon to report the local memory usage, and send it
+     * to rank=0 */
+    if (0 == orte_process_info.my_local_rank) {
+        sample();
     } else {
         /* now wait for notification */
         while (wait_for_release) {
@@ -157,60 +225,21 @@ int main(int argc, char* argv[])
 
     if (0 == rank) {
         fprintf(stderr, "\n\nSampling memory usage after MPI_Barrier\n");
-        OBJ_CONSTRUCT(&query, opal_list_t);
-        OBJ_CONSTRUCT(&response, opal_list_t);
-        q = OBJ_NEW(opal_pmix_query_t);
-        opal_list_append(&query, &q->super);
-        opal_argv_append_nosize(&q->keys, OPAL_PMIX_QUERY_MEMORY_USAGE);
-        /* qualify that we just want avg, min/max values reported */
-        kv = OBJ_NEW(opal_value_t);
-        kv->key = strdup(OPAL_PMIX_QUERY_REPORT_AVG);
-        kv->type = OPAL_BOOL;
-        kv->data.flag = true;
-        opal_list_append(&q->qualifiers, &kv->super);
-        kv = OBJ_NEW(opal_value_t);
-        kv->key = strdup(OPAL_PMIX_QUERY_REPORT_MINMAX);
-        kv->type = OPAL_BOOL;
-        kv->data.flag = true;
-        opal_list_append(&q->qualifiers, &kv->super);
-        /* issue the request */
-        wait_for_release = true;
-        opal_pmix.query(&query, qcbfunc, (void*)&response);
-        while (wait_for_release) {
-            usleep(10);
-        }
-        /* output the results */
-        OPAL_LIST_FOREACH(kv, &response, opal_value_t) {
-            fprintf(stderr, "\tResults: %s\n", kv->key);
-        }
-        OPAL_LIST_DESTRUCT(&response);
-        /* send the notification to release the other procs */
-        wait_for_release = true;
-        OBJ_CONSTRUCT(&response, opal_list_t);
-        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(&response, &kv->super);
-        if (OPAL_SUCCESS != opal_pmix.notify_event(MEMPROBE_RELEASE, NULL,
-                                                   OPAL_PMIX_RANGE_GLOBAL, &response,
-                                                   NULL, NULL)) {
-            fprintf(stderr, "Notify event failed\n");
-            exit(1);
-        }
-        while (wait_for_release) {
-            usleep(10);
+    }
+
+    if (0 == orte_process_info.my_local_rank) {
+        if (0 != rank) {
+            /* wait a little */
+            usleep(1000);
         }
-        OPAL_LIST_DESTRUCT(&response);
+        sample();
     } else {
         /* wait again while memory is sampled */
         while (wait_for_release) {
             usleep(10);
         }
     }
 
-    fprintf(stderr, "%d: FINALIZING\n", rank);
-    fflush(stderr);
     MPI_Finalize();
     return 0;
 }

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

@@ -10,7 +10,7 @@
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
  * Copyright (c) 2012      Los Alamos National Security, Inc.  All rights reserved.
- * Copyright (c) 2014-2016 Intel, Inc. All rights reserved.
+ * Copyright (c) 2014-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2015      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2016      Mellanox Technologies, Inc.
@@ -688,6 +688,7 @@ pmix_status_t pmix_bfrop_unpack_status(pmix_buffer_t *buffer, void *dest,
                 return PMIX_ERR_NOMEM;
             }
             if (PMIX_SUCCESS != (ret = pmix_bfrop_unpack_buffer(buffer, val->data.darray, &m, PMIX_DATA_ARRAY))) {
+                PMIX_ERROR_LOG(ret);
                 return ret;
             }
             break;
@@ -1274,6 +1275,9 @@ pmix_status_t pmix_bfrop_unpack_darray(pmix_buffer_t *buffer, void *dest,
             case PMIX_COMPRESSED_STRING:
                 nbytes = sizeof(pmix_byte_object_t);
                 break;
+            case PMIX_INFO:
+                nbytes = sizeof(pmix_info_t);
+                break;
             case PMIX_PERSIST:
                 nbytes = sizeof(pmix_persistence_t);
                 break;

opal/mca/pmix/pmix2x/pmix/src/common/pmix_query.c

@@ -1,6 +1,6 @@
 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2014-2016 Intel, Inc.  All rights reserved.
+ * Copyright (c) 2014-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2016      Mellanox Technologies, Inc.
  *                         All rights reserved.
  * Copyright (c) 2016      IBM Corporation.  All rights reserved.
@@ -78,7 +78,11 @@ static void query_cbfunc(struct pmix_peer_t *peer,
         cnt = results->ninfo;
         if (PMIX_SUCCESS != (rc = pmix_bfrop.unpack(buf, results->info, &cnt, PMIX_INFO))) {
             PMIX_ERROR_LOG(rc);
-            goto complete;
+            pmix_output(0, "TYPE: %d", results->info[0].value.type);
+            results->status = rc;
+            PMIX_INFO_FREE(results->info, results->ninfo);
+            results->info = NULL;
+            results->ninfo = 0;
         }
     }