Rework the threading in oob/tcp so that daemons (including mpirun) use multiple progress threads to get messages out to their children, and so that the oob/base uses a separate one to setup sends. This allows the daemon cmd processor to execute in parallel with relay of messages, which significantly reduces launch times at scale

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

Author: Ralph Castain

orte/mca/oob/base/base.h

@@ -11,6 +11,7 @@
  *                         All rights reserved.
  * Copyright (c) 2012-2013 Los Alamos National Security, LLC. All rights
  *                         reserved.
+ * Copyright (c) 2017      Intel, Inc.  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -54,13 +55,14 @@ OPAL_TIMING_DECLARE_EXT(ORTE_DECLSPEC, tm_oob)
  * Convenience Typedef
  */
 typedef struct {
+    opal_event_base_t *ev_base;
     char *include;
     char *exclude;
     opal_list_t components;
     opal_list_t actives;
     int max_uri_length;
     opal_hash_table_t peers;
-    bool use_module_threads;
+    int num_threads;
 #if OPAL_ENABLE_TIMING
     bool timing;
 #endif
@@ -119,7 +121,7 @@ ORTE_DECLSPEC void orte_oob_base_send_nb(int fd, short args, void *cbdata);
                             __FILE__, __LINE__);                        \
         cd = OBJ_NEW(orte_oob_send_t);                                  \
         cd->msg = (m);                                                  \
-        opal_event_set(orte_event_base, &cd->ev, -1,                    \
+        opal_event_set(orte_oob_base.ev_base, &cd->ev, -1,              \
                        OPAL_EV_WRITE,                                   \
                        orte_oob_base_send_nb, cd);                      \
         opal_event_set_priority(&cd->ev, ORTE_MSG_PRI);                 \
@@ -173,7 +175,7 @@ OBJ_CLASS_DECLARATION(mca_oob_uri_req_t);
         mca_oob_uri_req_t *rq;                                  \
         rq = OBJ_NEW(mca_oob_uri_req_t);                        \
         rq->uri = strdup((u));                                  \
-        opal_event_set(orte_event_base, &(rq)->ev, -1,          \
+        opal_event_set(orte_oob_base.ev_base, &(rq)->ev, -1,    \
                        OPAL_EV_WRITE,                           \
                        orte_oob_base_set_addr, (rq));           \
         opal_event_set_priority(&(rq)->ev, ORTE_MSG_PRI);       \
@@ -193,4 +195,3 @@ ORTE_DECLSPEC void orte_oob_base_ft_event(int fd, short args, void *cbdata);
 
 END_C_DECLS
 #endif
-

orte/mca/oob/base/oob_base_frame.c

@@ -29,6 +29,7 @@
 
 #include "opal/class/opal_bitmap.h"
 #include "orte/mca/mca.h"
+#include "opal/runtime/opal_progress_threads.h"
 #include "opal/util/output.h"
 #include "opal/mca/base/base.h"
 
@@ -53,19 +54,20 @@
 orte_oob_base_t orte_oob_base = {0};
 OPAL_TIMING_DECLARE(tm_oob)
 
+
 static int orte_oob_base_register(mca_base_register_flag_t flags)
 {
     if (ORTE_PROC_IS_APP || ORTE_PROC_IS_TOOL) {
-        orte_oob_base.use_module_threads = false;
+        orte_oob_base.num_threads = 0;
     } else {
-        orte_oob_base.use_module_threads = true;
+        orte_oob_base.num_threads = 8;
     }
-    (void)mca_base_var_register("orte", "oob", "base", "enable_module_progress_threads",
-                                "Whether to independently progress OOB messages for each interface",
+    (void)mca_base_var_register("orte", "oob", "base", "num_progress_threads",
+                                "Number of independent progress OOB messages for each interface",
                                 MCA_BASE_VAR_TYPE_BOOL, NULL, 0, 0,
                                 OPAL_INFO_LVL_9,
                                 MCA_BASE_VAR_SCOPE_READONLY,
-                                &orte_oob_base.use_module_threads);
+                                &orte_oob_base.num_threads);
 
 #if OPAL_ENABLE_TIMING
     /* Detailed timing setup */
@@ -107,6 +109,11 @@ static int orte_oob_base_close(void)
 
     OBJ_DESTRUCT(&orte_oob_base.peers);
 
+    if (ORTE_PROC_IS_APP || ORTE_PROC_IS_TOOL) {
+        opal_progress_thread_finalize(NULL);
+    } else {
+        opal_progress_thread_finalize("OOB-BASE");
+    }
 
     OPAL_TIMING_EVENT((&tm_oob, "Finish"));
     OPAL_TIMING_REPORT(orte_oob_base.timing, &tm_oob);
@@ -126,6 +133,13 @@ static int orte_oob_base_open(mca_base_open_flag_t flags)
     opal_hash_table_init(&orte_oob_base.peers, 128);
     OBJ_CONSTRUCT(&orte_oob_base.actives, opal_list_t);
 
+    if (ORTE_PROC_IS_APP || ORTE_PROC_IS_TOOL) {
+        orte_oob_base.ev_base = opal_progress_thread_init(NULL);
+    } else {
+        orte_oob_base.ev_base = opal_progress_thread_init("OOB-BASE");
+    }
+
+
 #if OPAL_ENABLE_FT_CR == 1
     /* register the FT events callback */
     orte_state.add_job_state(ORTE_JOB_STATE_FT_CHECKPOINT, orte_oob_base_ft_event, ORTE_ERROR_PRI);

orte/mca/oob/tcp/Makefile.am

@@ -12,7 +12,7 @@
 # Copyright (c) 2010      Cisco Systems, Inc.  All rights reserved.
 # Copyright (c) 2012-2013 Los Alamos National Security, LLC.
 #                         All rights reserved
-# Copyright (c) 2014      Intel, Inc.  All rights reserved.
+# Copyright (c) 2014-2017 Intel, Inc.  All rights reserved.
 # $COPYRIGHT$
 #
 # Additional copyrights may follow
@@ -32,7 +32,6 @@ sources = \
           oob_tcp_sendrecv.h \
           oob_tcp_hdr.h \
           oob_tcp_peer.h \
-          oob_tcp_ping.h \
           oob_tcp.c \
           oob_tcp_listener.c \
           oob_tcp_common.c \
@@ -59,4 +58,3 @@ mca_oob_tcp_la_LDFLAGS = -module -avoid-version
 noinst_LTLIBRARIES = $(component_noinst)
 libmca_oob_tcp_la_SOURCES = $(sources)
 libmca_oob_tcp_la_LDFLAGS = -module -avoid-version
-