Complete the memprobe support. This provides a new scaling tool called "mpi_memprobe" that samples the memory footprint of the local daemon and the client procs, and then reports the results. The output contains the footprint of the daemon on each node, plus the average footprint of the client procs on that node.

Samples are taken after MPI_Init, and then again after MPI_Barrier. This allows the user to see memory consumption caused by add_procs, as well as any modex contribution from forming connections if pmix_base_async_modex is given.

Using the probe simply involves executing it via mpirun, with however many copies you want per node. Example:

$ mpirun -npernode 2 ./mpi_memprobe
Sampling memory usage after MPI_Init
Data for node rhc001
	Daemon: 12.483398
	Client: 6.514648

Data for node rhc002
	Daemon: 11.865234
	Client: 4.643555

Sampling memory usage after MPI_Barrier
Data for node rhc001
	Daemon: 12.520508
	Client: 6.576660

Data for node rhc002
	Daemon: 11.879883
	Client: 4.703125

Note that the client value on node rhc001 is larger - this is where rank=0 is housed, and apparently it gets a larger footprint for some reason.

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

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;
         }
     }