Refactor show_help() to use the PMIx_Log() api.

- Port over the de-dupping code from prrte.
- Add some additional keys to support a request
  to de-dup logging messages.
- Remove locks from the high level PMIx_Log() api's - they
  don't seem necessary, and make it very easy to deadlock
  when the current thread already holds the global lock.
- Likewise - use PMIx_Log_nb() instead of PMIx_Log() - the
  non-blocking counterpart will allow libevent to trigger the
  callbacks on its own time. Otherwise, in the case of a
  show_help() message from prted that happens before the main thread
  hits the event loop, it will hang forever.
- We can avoid deadlock in show_help calls made prior to
  completing client/server/tool "init" by simply leaving
  the show_help system disabled until we get far enough
  along, and then explicitly "enabling" it. We also never
  call a PMIx API from inside the code base as that can
  lead to loopbacks that deadlock.

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

Co-authored-by: Ralph Castain <[email protected]>

Author: awlauria

include/pmix_common.h.in

@@ -674,6 +674,10 @@ typedef uint32_t pmix_rank_t;
 #define PMIX_LOG_XML_OUTPUT                 "pmix.log.xml"          // (bool) print the output stream in xml format
 #define PMIX_LOG_ONCE                       "pmix.log.once"         // (bool) only log this once with whichever channel can first support it
 #define PMIX_LOG_MSG                        "pmix.log.msg"          // (pmix_byte_object_t) message blob to be sent somewhere
+#define PMIX_LOG_KEY                        "pmix.log.key"          // (char*) key to a logging message
+#define PMIX_LOG_VAL                        "pmix.log.val"          // (char*) value to a logging message
+#define PMIX_LOG_AGG                        "pmix.log.agg"          // (bool) Whether to aggregate and prevent duplicate logging messages
+                                                                    //        based on key value pairs.
 
 #define PMIX_LOG_EMAIL                      "pmix.log.email"        // (pmix_data_array_t*) log via email based on array of pmix_info_t
                                                                     //         containing directives

src/client/pmix_client.c

@@ -78,6 +78,7 @@ static pmix_status_t pmix_init_result = PMIX_ERR_INIT;
 #include "src/util/pmix_name_fns.h"
 #include "src/util/pmix_output.h"
 #include "src/util/pmix_printf.h"
+#include "src/util/pmix_show_help.h"
 
 #include "pmix_client_ops.h"
 
@@ -860,6 +861,8 @@ PMIX_EXPORT pmix_status_t PMIx_Init(pmix_proc_t *proc, pmix_info_t info[], size_
         PMIX_RELEASE_THREAD(&pmix_global_lock);
         return rc;
     }
+    // enable show_help subsystem
+    pmix_show_help_enabled = true;
     PMIX_RELEASE_THREAD(&pmix_global_lock);
 
     /* look for a debugger attach key */

src/common/pmix_log.c

@@ -69,13 +69,9 @@ PMIX_EXPORT pmix_status_t PMIx_Log(const pmix_info_t data[], size_t ndata,
     pmix_cb_t cb;
     pmix_status_t rc;
 
-    PMIX_ACQUIRE_THREAD(&pmix_global_lock);
-
     if (pmix_globals.init_cntr <= 0) {
-        PMIX_RELEASE_THREAD(&pmix_global_lock);
         return PMIX_ERR_INIT;
     }
-    PMIX_RELEASE_THREAD(&pmix_global_lock);
 
     pmix_output_verbose(2, pmix_plog_base_framework.framework_output, "%s pmix:log",
                         PMIX_NAME_PRINT(&pmix_globals.myid));
@@ -115,37 +111,77 @@ static void localcbfunc(pmix_status_t status, void *cbdata)
     PMIX_RELEASE(cd);
 }
 
+void pmix_log_local_op(int sd, short args, void *cbdata_)
+{
+    pmix_shift_caddy_t *qd = (pmix_shift_caddy_t *) cbdata_;
+    pmix_info_t *data = qd->info;
+    size_t ndata = qd->ninfo;
+    pmix_info_t *directives = qd->directives;
+    size_t ndirs = qd->ndirs;
+    pmix_op_cbfunc_t cbfunc = qd->cbfunc.opcbfn;
+    void *cbdata = qd->cbdata;
+    pmix_proc_t *source = qd->proc;
+    pmix_status_t rc;
+    pmix_shift_caddy_t *cd;
+    size_t n;
+    PMIX_HIDE_UNUSED_PARAMS(sd, args);
+
+    /* if no recorded source was found, then we must be it */
+    if (NULL == source) {
+        source = &pmix_globals.myid;
+        cd = PMIX_NEW(pmix_shift_caddy_t);
+        cd->cbfunc.opcbfn = cbfunc;
+        cd->cbdata = cbdata;
+        cd->ndirs = ndirs + 1;
+        PMIX_INFO_CREATE(cd->directives, cd->ndirs);
+        for (n = 0; n < ndirs; n++) {
+            PMIX_INFO_XFER(&cd->directives[n], (pmix_info_t *) &directives[n]);
+        }
+        PMIX_INFO_LOAD(&cd->directives[ndirs], PMIX_LOG_SOURCE, source, PMIX_PROC);
+        rc = pmix_plog.log(source, data, ndata, cd->directives, cd->ndirs, localcbfunc, cd);
+        if (PMIX_SUCCESS != rc) {
+            PMIX_INFO_FREE(cd->directives, cd->ndirs);
+            PMIX_RELEASE(cd);
+        }
+    } else if (PMIX_CHECK_PROCID(source, &pmix_globals.myid)) {
+        /* if I am the recorded source, then this is a re-submission of
+         * something that got "upcalled" by a prior call. In this case,
+         * we return a "not supported" error as clearly we couldn't
+         * handle it, and neither could our host */
+        rc = PMIX_ERR_NOT_SUPPORTED;
+    } else {
+        /* call down to process the request - the various components
+         * will thread shift as required */
+        rc = pmix_plog.log(source, data, ndata, directives, ndirs, cbfunc, cbdata);
+    }
+}
+
 PMIX_EXPORT pmix_status_t PMIx_Log_nb(const pmix_info_t data[], size_t ndata,
                                       const pmix_info_t directives[], size_t ndirs,
                                       pmix_op_cbfunc_t cbfunc, void *cbdata)
 
 {
-    pmix_shift_caddy_t *cd;
     pmix_cmd_t cmd = PMIX_LOG_CMD;
     pmix_buffer_t *msg;
-    pmix_status_t rc;
-    size_t n;
+    pmix_status_t rc = PMIX_SUCCESS;
     time_t timestamp = 0;
     pmix_proc_t *source = NULL;
-
-    PMIX_ACQUIRE_THREAD(&pmix_global_lock);
+    pmix_shift_caddy_t *cd;
 
     pmix_output_verbose(2, pmix_globals.debug_output, "pmix:log non-blocking");
 
     if (pmix_globals.init_cntr <= 0) {
-        PMIX_RELEASE_THREAD(&pmix_global_lock);
         return PMIX_ERR_INIT;
     }
 
     if (0 == ndata || NULL == data) {
-        PMIX_RELEASE_THREAD(&pmix_global_lock);
         return PMIX_ERR_BAD_PARAM;
     }
 
     /* check the directives - if they requested a timestamp, then
      * get the time, also look for a source */
     if (NULL != directives) {
-        for (n = 0; n < ndirs; n++) {
+        for (size_t n = 0; n < ndirs; n++) {
             if (0 == strncmp(directives[n].key, PMIX_LOG_GENERATE_TIMESTAMP, PMIX_MAX_KEYLEN)) {
                 if (PMIX_INFO_TRUE(&directives[n])) {
                     /* pickup the timestamp */
@@ -162,10 +198,8 @@ PMIX_EXPORT pmix_status_t PMIx_Log_nb(const pmix_info_t data[], size_t ndata,
     if (!PMIX_PEER_IS_SERVER(pmix_globals.mypeer) && !PMIX_PEER_IS_LAUNCHER(pmix_globals.mypeer)) {
         /* if we aren't connected, don't attempt to send */
         if (!pmix_globals.connected) {
-            PMIX_RELEASE_THREAD(&pmix_global_lock);
             return PMIX_ERR_UNREACH;
         }
-        PMIX_RELEASE_THREAD(&pmix_global_lock);
 
         /* if we are not a server, then relay this request to the server */
         cd = PMIX_NEW(pmix_shift_caddy_t);
@@ -234,38 +268,16 @@ PMIX_EXPORT pmix_status_t PMIx_Log_nb(const pmix_info_t data[], size_t ndata,
         }
         return rc;
     }
-    PMIX_RELEASE_THREAD(&pmix_global_lock);
 
-    /* if no recorded source was found, then we must be it */
-    if (NULL == source) {
-        source = &pmix_globals.myid;
-        cd = PMIX_NEW(pmix_shift_caddy_t);
-        cd->cbfunc.opcbfn = cbfunc;
-        cd->cbdata = cbdata;
-        cd->ndirs = ndirs + 1;
-        PMIX_INFO_CREATE(cd->directives, cd->ndirs);
-        for (n = 0; n < ndirs; n++) {
-            PMIX_INFO_XFER(&cd->directives[n], (pmix_info_t *) &directives[n]);
-        }
-        PMIX_INFO_LOAD(&cd->directives[ndirs], PMIX_LOG_SOURCE, source, PMIX_PROC);
-        /* call down to process the request - the various components
-         * will thread shift as required */
-        rc = pmix_plog.log(source, data, ndata, cd->directives, cd->ndirs, localcbfunc, cd);
-        if (PMIX_SUCCESS != rc) {
-            PMIX_INFO_FREE(cd->directives, cd->ndirs);
-            PMIX_RELEASE(cd);
-        }
-    } else if (PMIX_CHECK_PROCID(source, &pmix_globals.myid)) {
-        /* if I am the recorded source, then this is a re-submission of
-         * something that got "upcalled" by a prior call. In this case,
-         * we return a "not supported" error as clearly we couldn't
-         * handle it, and neither could our host */
-        rc = PMIX_ERR_NOT_SUPPORTED;
-    } else {
-        /* call down to process the request - the various components
-         * will thread shift as required */
-        rc = pmix_plog.log(source, data, ndata, directives, ndirs, cbfunc, cbdata);
-    }
+    cd = PMIX_NEW(pmix_shift_caddy_t);
+    cd->info = (pmix_info_t *) data;
+    cd->ninfo = ndata;
+    cd->directives = (pmix_info_t *) directives;
+    cd->ndirs = ndirs;
+    cd->cbfunc.opcbfn = cbfunc;
+    cd->cbdata = cbdata;
+    cd->proc = source;
+    PMIX_THREADSHIFT(cd, pmix_log_local_op);
 
     return rc;
 }

src/include/pmix_globals.c

@@ -295,6 +295,7 @@ static void scon(pmix_shift_caddy_t *p)
     p->codes = NULL;
     p->ncodes = 0;
     p->peer = NULL;
+    p->proc = NULL;
     p->pname.nspace = NULL;
     p->pname.rank = PMIX_RANK_UNDEF;
     p->data = NULL;

src/include/pmix_globals.h

@@ -369,7 +369,9 @@ typedef struct {
     pmix_proc_t *targets;
     size_t ntargets;
     pmix_info_t *info;
+    pmix_info_t *dirs;
     size_t ninfo;
+    size_t ndirs;
     pmix_list_t results;
     size_t nreplies;
     size_t nrequests;
@@ -442,6 +444,7 @@ typedef struct {
     pmix_status_t *codes;
     size_t ncodes;
     pmix_name_t pname;
+    pmix_proc_t *proc;
     pmix_peer_t *peer;
     const char *data;
     size_t ndata;
@@ -638,6 +641,8 @@ PMIX_EXPORT extern pmix_lock_t pmix_global_lock;
 PMIX_EXPORT extern const char* PMIX_PROXY_VERSION;
 PMIX_EXPORT extern const char* PMIX_PROXY_BUGREPORT;
 
+PMIX_EXPORT void pmix_log_local_op(int sd, short args, void *cbdata_);
+
 static inline bool pmix_check_node_info(const char *key)
 {
     char *keys[] = {PMIX_HOSTNAME,

src/mca/plog/base/plog_base_stubs.c

@@ -18,6 +18,7 @@
 #include "src/class/pmix_list.h"
 #include "src/server/pmix_server_ops.h"
 #include "src/util/pmix_error.h"
+#include "src/util/pmix_show_help.h"
 
 #include "src/mca/plog/base/base.h"
 
@@ -77,6 +78,8 @@ pmix_status_t pmix_plog_base_log(const pmix_proc_t *source, const pmix_info_t da
     pmix_mycount_t *mycount;
     pmix_list_t channels;
     bool all_complete = true;
+    char *key = NULL, *val = NULL;
+    bool agg = true;  // default to aggregating show_help messages
 
     if (!pmix_plog_globals.initialized) {
         return PMIX_ERR_INIT;
@@ -108,7 +111,24 @@ pmix_status_t pmix_plog_base_log(const pmix_proc_t *source, const pmix_info_t da
         for (n = 0; n < ndirs; n++) {
             if (PMIX_CHECK_KEY(&directives[n], PMIX_LOG_ONCE)) {
                 logonce = PMIX_INFO_TRUE(&directives[n]);
-                break;
+            }
+            else if (PMIX_CHECK_KEY(&directives[n], PMIX_LOG_AGG)) {
+                    agg = PMIX_INFO_TRUE(&directives[n]);
+            }
+            else if (PMIX_CHECK_KEY(&directives[n], PMIX_LOG_KEY)) {
+                key = directives[n].value.data.string;
+            }
+            else if (PMIX_CHECK_KEY(&directives[n], PMIX_LOG_VAL)) {
+                val = directives[n].value.data.string;
+            }
+        }
+        if (agg && NULL != key && NULL != val) {
+            if (PMIX_SUCCESS == pmix_help_check_dups(key, val)) {
+                for (k = 0; k < ndata; k++) {
+                    // This is a dup and has been tracked as such,
+                    // mark this as complete so we don't log it again.
+                    PMIX_INFO_OP_COMPLETED(&data[k]);
+                }
             }
         }
     }
@@ -162,7 +182,10 @@ pmix_status_t pmix_plog_base_log(const pmix_proc_t *source, const pmix_info_t da
         PMIX_DESTRUCT(&channels);
         PMIX_RELEASE(mycount);
         PMIX_RELEASE_THREAD(&pmix_plog_globals.lock);
-        return PMIX_SUCCESS;
+
+        // Don't return PMIX_SUCCESS here, or else the called
+        // will expect the cbfunc to be executed (which it won't be.).
+        return PMIX_OPERATION_SUCCEEDED;
     }
     PMIX_ACQUIRE_THREAD(&mycount->lock);
     PMIX_LIST_FOREACH (active, &channels, pmix_plog_base_active_module_t) {

src/runtime/pmix_params.c

@@ -114,6 +114,7 @@ pmix_status_t pmix_register_params(void)
                                       PMIX_MCA_BASE_VAR_TYPE_BOOL,
                                       &pmix_suppress_missing_data_warning);
 
+
     /****   CLIENT: VERBOSE OUTPUT PARAMS   ****/
     (void) pmix_mca_base_var_register("pmix", "pmix", "client", "get_verbose",
                                       "Verbosity for client get operations",

src/server/pmix_server.c

@@ -837,6 +837,8 @@ PMIX_EXPORT pmix_status_t PMIx_server_init(pmix_server_module_t *module, pmix_in
     }
 
     ++pmix_globals.init_cntr;
+    // enable show_help subsystem
+    pmix_show_help_enabled = true;
     PMIX_RELEASE_THREAD(&pmix_global_lock);
 
     /* register a handler to catch/aggregate PMIX_EVENT_WAITING_FOR_NOTIFY

src/tool/pmix_tool.c

@@ -943,6 +943,8 @@ PMIX_EXPORT int PMIx_tool_init(pmix_proc_t *proc, pmix_info_t info[], size_t nin
         }
         PMIX_DESTRUCT(&cb);
     }
+    // enable show_help subsystem
+    pmix_show_help_enabled = true;
     PMIX_RELEASE_THREAD(&pmix_global_lock);
 
     /* if we are acting as a server, then start listening */