Merge pull request #3143 from rhc54/topic/odls

Enable parallel fork/exec of local procs by providing the option of multiple odls progress threads

Author: Ralph Castain

orte/mca/iof/base/iof_base_setup.c

@@ -219,11 +219,6 @@ orte_iof_base_setup_parent(const orte_process_name_t* name,
 {
     int ret;
 
-    close(opts->p_stdin[0]);
-    close(opts->p_stdout[1]);
-    close(opts->p_stderr[1]);
-    close(opts->p_internal[1]);
-
     /* connect stdin endpoint */
     if (opts->connect_stdin) {
         /* and connect the pty to stdin */

orte/mca/odls/base/odls_base_default_fns.c

@@ -15,6 +15,7 @@
  *                         All rights reserved.
  * Copyright (c) 2014-2015 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2017      Intel, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -32,6 +33,7 @@
 #include "orte/mca/mca.h"
 #include "opal/mca/base/base.h"
 #include "opal/mca/hwloc/hwloc-internal.h"
+#include "opal/runtime/opal_progress_threads.h"
 #include "opal/util/output.h"
 #include "opal/util/path.h"
 #include "opal/util/argv.h"
@@ -76,6 +78,14 @@ static int orte_odls_base_register(mca_base_register_flag_t flags)
                                  MCA_BASE_VAR_SCOPE_READONLY,
                                  &orte_odls_globals.timeout_before_sigkill);
 
+    orte_odls_globals.num_threads = 0;
+    (void) mca_base_var_register("orte", "odls", "base", "num_threads",
+                                 "Number of threads to use for spawning local procs",
+                                 MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
+                                 OPAL_INFO_LVL_9,
+                                 MCA_BASE_VAR_SCOPE_READONLY,
+                                 &orte_odls_globals.num_threads);
+
     return ORTE_SUCCESS;
 }
 
@@ -99,6 +109,15 @@ static int orte_odls_base_close(void)
     }
     OBJ_RELEASE(orte_local_children);
 
+    if (0 < orte_odls_globals.num_threads) {
+        /* stop the progress threads */
+        for (i=0; NULL != orte_odls_globals.ev_threads[i]; i++) {
+            opal_progress_thread_finalize(orte_odls_globals.ev_threads[i]);
+        }
+    }
+    free(orte_odls_globals.ev_bases);
+    opal_argv_free(orte_odls_globals.ev_threads);
+
     return mca_base_framework_components_close(&orte_odls_base_framework, NULL);
 }
 
@@ -174,6 +193,25 @@ static int orte_odls_base_open(mca_base_open_flag_t flags)
         opal_argv_append_nosize(&orte_odls_globals.xtermcmd, "-e");
     }
 
+    /* setup the pool of worker threads */
+    orte_odls_globals.ev_threads = NULL;
+    orte_odls_globals.next_base = 0;
+    if (0 == orte_odls_globals.num_threads) {
+        orte_odls_globals.ev_bases = (opal_event_base_t**)malloc(sizeof(opal_event_base_t*));
+        /* use the default event base */
+       orte_odls_globals.ev_bases[0] = orte_event_base;
+    } else {
+        orte_odls_globals.ev_bases =
+            (opal_event_base_t**)malloc(orte_odls_globals.num_threads * sizeof(opal_event_base_t*));
+        for (i=0; i < orte_odls_globals.num_threads; i++) {
+            asprintf(&tmp, "ORTE-ODLS-%d", i);
+            orte_odls_globals.ev_bases[i] = opal_progress_thread_init(tmp);
+            opal_argv_append_nosize(&orte_odls_globals.ev_threads, tmp);
+            free(tmp);
+        }
+
+    }
+
      /* Open up all available components */
     return mca_base_framework_components_open(&orte_odls_base_framework, flags);
 }
@@ -197,3 +235,11 @@ OBJ_CLASS_INSTANCE(orte_odls_launch_local_t,
                    opal_object_t,
                    launch_local_const,
                    launch_local_dest);
+
+static void sccon(orte_odls_spawn_caddy_t *p)
+{
+    memset(&p->opts, 0, sizeof(orte_iof_base_io_conf_t));
+}
+OBJ_CLASS_INSTANCE(orte_odls_spawn_caddy_t,
+                   opal_object_t,
+                   sccon, NULL);

orte/mca/odls/base/odls_base_frame.c

@@ -12,7 +12,7 @@
  * Copyright (c) 2011      Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2011      Los Alamos National Security, LLC.  All rights
  *                         reserved.
- * Copyright (c) 2016      Intel, Inc. All rights reserved.
+ * Copyright (c) 2016-2017 Intel, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -36,6 +36,7 @@
 #include "opal/class/opal_bitmap.h"
 #include "opal/dss/dss_types.h"
 
+#include "orte/mca/iof/base/iof_base_setup.h"
 #include "orte/mca/rml/rml_types.h"
 #include "orte/runtime/orte_globals.h"
 
@@ -56,11 +57,15 @@ typedef struct {
     opal_list_t xterm_ranks;
     /* the xterm cmd to be used */
     char **xtermcmd;
+    /* thread pool */
+    int num_threads;
+    opal_event_base_t **ev_bases;   // event base array for progress threads
+    char** ev_threads;              // event progress thread names
+    int next_base;                  // counter to load-level thread use
 } orte_odls_globals_t;
 
 ORTE_DECLSPEC extern orte_odls_globals_t orte_odls_globals;
 
-
 /*
  * Default functions that are common to most environments - can
  * be overridden by specific environments if they need something
@@ -74,11 +79,27 @@ ORTE_DECLSPEC int
 orte_odls_base_default_construct_child_list(opal_buffer_t *data,
                                             orte_jobid_t *job);
 
+ORTE_DECLSPEC void orte_odls_base_spawn_proc(int fd, short sd, void *cbdata);
+
 /* define a function that will fork a local proc */
-typedef int (*orte_odls_base_fork_local_proc_fn_t)(orte_app_context_t *context,
-                                                   orte_proc_t *child,
+typedef int (*orte_odls_base_fork_local_proc_fn_t)(orte_proc_t *child,
+                                                   char *app, char **argv,
                                                    char **environ_copy,
-                                                   orte_job_t *jdata);
+                                                   orte_job_t *jdata,
+                                                   orte_iof_base_io_conf_t opts);
+
+/* define an object for fork/exec the local proc */
+typedef struct {
+    opal_object_t super;
+    opal_event_t ev;
+    orte_job_t *jdata;
+    orte_app_context_t *app;
+    orte_proc_t *child;
+    bool index_argv;
+    orte_iof_base_io_conf_t opts;
+    orte_odls_base_fork_local_proc_fn_t fork_local;
+} orte_odls_spawn_caddy_t;
+OBJ_CLASS_DECLARATION(orte_odls_spawn_caddy_t);
 
 /* define an object for starting local launch */
 typedef struct {

orte/mca/odls/base/odls_private.h

@@ -15,7 +15,7 @@
  * Copyright (c) 2010      IBM Corporation.  All rights reserved.
  * Copyright (c) 2011-2013 Los Alamos National Security, LLC.  All rights
  *                         reserved.
- * Copyright (c) 2013-2016 Intel, Inc. All rights reserved
+ * Copyright (c) 2013-2017 Intel, Inc. All rights reserved.
  *
  * $COPYRIGHT$
  *
@@ -144,8 +144,9 @@ static int orte_odls_default_restart_proc(orte_proc_t *child);
 static void send_error_show_help(int fd, int exit_status,
                                  const char *file, const char *topic, ...)
     __opal_attribute_noreturn__;
-static int do_child(orte_app_context_t* context,
-                    orte_proc_t *child,
+
+static int do_child(orte_proc_t *child,
+                    char *cmd, char **argv,
                     char **environ_copy,
                     orte_job_t *jobdat, int write_fd,
                     orte_iof_base_io_conf_t opts)
@@ -318,16 +319,15 @@ static int close_open_file_descriptors(int write_fd,
     return ORTE_SUCCESS;
 }
 
-static int do_child(orte_app_context_t* context,
-                    orte_proc_t *child,
+static int do_child(orte_proc_t *child,
+                    char *app, char **argv,
                     char **environ_copy,
                     orte_job_t *jobdat, int write_fd,
                     orte_iof_base_io_conf_t opts)
 {
-    int i, rc;
+    int i;
     sigset_t sigs;
     long fd, fdmax = sysconf(_SC_OPEN_MAX);
-    char *param, *msg;
 
 #if HAVE_SETPGID
     /* Set a new process group for this child, so that any
@@ -359,7 +359,7 @@ static int do_child(orte_app_context_t* context,
                 send_error_show_help(write_fd, 1,
                                      "help-orte-odls-default.txt",
                                      "iof setup failed",
-                                     orte_process_info.nodename, context->app);
+                                     orte_process_info.nodename, app);
                 /* Does not return */
             }
         }
@@ -384,18 +384,6 @@ static int do_child(orte_app_context_t* context,
         close(fdnull);
     }
 
-    /* if the user requested it, set the system resource limits */
-    if (OPAL_SUCCESS != (rc = opal_util_init_sys_limits(&msg))) {
-        send_error_show_help(write_fd, 1, "help-orte-odls-default.txt",
-                             "set limit",
-                             orte_process_info.nodename, context->app,
-                             __FILE__, __LINE__, msg);
-    }
-    /* ensure we only do this once */
-    (void) mca_base_var_env_name("opal_set_max_sys_limits", &param);
-    opal_unsetenv(param, &environ_copy);
-    free(param);
-
     /* close all open file descriptors w/ exception of stdin/stdout/stderr,
        the pipe used for the IOF INTERNAL messages, and the pipe up to
        the parent. */
@@ -408,10 +396,10 @@ static int do_child(orte_app_context_t* context,
         }
     }
 
-    if (context->argv == NULL) {
-        context->argv = malloc(sizeof(char*)*2);
-        context->argv[0] = strdup(context->app);
-        context->argv[1] = NULL;
+    if (argv == NULL) {
+        argv = malloc(sizeof(char*)*2);
+        argv[0] = strdup(app);
+        argv[1] = NULL;
     }
 
     /* Set signal handlers back to the default.  Do this close to
@@ -436,16 +424,16 @@ static int do_child(orte_app_context_t* context,
 
     /* Exec the new executable */
 
-    execve(context->app, context->argv, environ_copy);
+    execve(app, argv, environ_copy);
     send_error_show_help(write_fd, 1,
                          "help-orte-odls-default.txt", "execve error",
-                         orte_process_info.nodename, context->app, strerror(errno));
+                         orte_process_info.nodename, app, strerror(errno));
     /* Does not return */
 }
 
 
-static int do_parent(orte_app_context_t* context,
-                     orte_proc_t *child,
+static int do_parent(orte_proc_t *child,
+                     char *app, char **argv,
                      char **environ_copy,
                      orte_job_t *jobdat, int read_fd,
                      orte_iof_base_io_conf_t opts)
@@ -454,19 +442,10 @@ static int do_parent(orte_app_context_t* context,
     orte_odls_pipe_err_msg_t msg;
     char file[ORTE_ODLS_MAX_FILE_LEN + 1], topic[ORTE_ODLS_MAX_TOPIC_LEN + 1], *str = NULL;
 
-    if (NULL != child && ORTE_FLAG_TEST(jobdat, ORTE_JOB_FLAG_FORWARD_OUTPUT)) {
-        /* connect endpoints IOF */
-        rc = orte_iof_base_setup_parent(&child->name, &opts);
-        if (ORTE_SUCCESS != rc) {
-            ORTE_ERROR_LOG(rc);
-            close(read_fd);
-
-            if (NULL != child) {
-                child->state = ORTE_PROC_STATE_UNDEF;
-            }
-            return rc;
-        }
-    }
+    close(opts.p_stdin[0]);
+    close(opts.p_stdout[1]);
+    close(opts.p_stderr[1]);
+    close(opts.p_internal[1]);
 
     /* Block reading a message from the pipe */
     while (1) {
@@ -503,7 +482,7 @@ static int do_parent(orte_app_context_t* context,
             if (OPAL_SUCCESS != rc) {
                 orte_show_help("help-orte-odls-default.txt", "syscall fail",
                                true,
-                               orte_process_info.nodename, context->app,
+                               orte_process_info.nodename, app,
                                "opal_fd_read", __FILE__, __LINE__);
                 if (NULL != child) {
                     child->state = ORTE_PROC_STATE_UNDEF;
@@ -517,7 +496,7 @@ static int do_parent(orte_app_context_t* context,
             if (OPAL_SUCCESS != rc) {
                 orte_show_help("help-orte-odls-default.txt", "syscall fail",
                                true,
-                               orte_process_info.nodename, context->app,
+                               orte_process_info.nodename, app,
                                "opal_fd_read", __FILE__, __LINE__);
                 if (NULL != child) {
                     child->state = ORTE_PROC_STATE_UNDEF;
@@ -531,7 +510,7 @@ static int do_parent(orte_app_context_t* context,
             if (NULL == str) {
                 orte_show_help("help-orte-odls-default.txt", "syscall fail",
                                true,
-                               orte_process_info.nodename, context->app,
+                               orte_process_info.nodename, app,
                                "opal_fd_read", __FILE__, __LINE__);
                 if (NULL != child) {
                     child->state = ORTE_PROC_STATE_UNDEF;
@@ -580,39 +559,16 @@ static int do_parent(orte_app_context_t* context,
 /**
  *  Fork/exec the specified processes
  */
-static int odls_default_fork_local_proc(orte_app_context_t* context,
-                                        orte_proc_t *child,
+static int odls_default_fork_local_proc(orte_proc_t *child,
+                                        char *app,
+                                        char **argv,
                                         char **environ_copy,
-                                        orte_job_t *jobdat)
+                                        orte_job_t *jobdat,
+                                        orte_iof_base_io_conf_t opts)
 {
-    orte_iof_base_io_conf_t opts = {0};
-    int rc, p[2];
+    int p[2];
     pid_t pid;
 
-    if (NULL != child) {
-        /* should pull this information from MPIRUN instead of going with
-         default */
-        opts.usepty = OPAL_ENABLE_PTY_SUPPORT;
-
-        /* do we want to setup stdin? */
-        if (NULL != child &&
-            (jobdat->stdin_target == ORTE_VPID_WILDCARD ||
-             child->name.vpid == jobdat->stdin_target)) {
-            opts.connect_stdin = true;
-        } else {
-            opts.connect_stdin = false;
-        }
-
-        if (ORTE_SUCCESS != (rc = orte_iof_base_setup_prefork(&opts))) {
-            ORTE_ERROR_LOG(rc);
-            if (NULL != child) {
-                child->state = ORTE_PROC_STATE_FAILED_TO_START;
-                child->exit_code = rc;
-            }
-            return rc;
-        }
-    }
-
     /* A pipe is used to communicate between the parent and child to
        indicate whether the exec ultimately succeeded or failed.  The
        child sets the pipe to be close-on-exec; the child only ever
@@ -647,12 +603,12 @@ static int odls_default_fork_local_proc(orte_app_context_t* context,
 
     if (pid == 0) {
         close(p[0]);
-        do_child(context, child, environ_copy, jobdat, p[1], opts);
+        do_child(child, app, argv, environ_copy, jobdat, p[1], opts);
         /* Does not return */
     }
 
     close(p[1]);
-    return do_parent(context, child, environ_copy, jobdat, p[0], opts);
+    return do_parent(child, app, argv, environ_copy, jobdat, p[0], opts);
 }
 
 

orte/mca/odls/default/odls_default_module.c

@@ -73,7 +73,8 @@ ORTE_DECLSPEC int orte_schizo_base_setup_fork(orte_job_t *jdata,
                                               orte_app_context_t *context);
 ORTE_DECLSPEC int orte_schizo_base_setup_child(orte_job_t *jobdat,
                                                orte_proc_t *child,
-                                               orte_app_context_t *app);
+                                               orte_app_context_t *app,
+                                               char ***env);
 ORTE_DECLSPEC orte_schizo_launch_environ_t orte_schizo_base_check_launch_environment(void);
 ORTE_DECLSPEC long orte_schizo_base_get_remaining_time(void);
 ORTE_DECLSPEC void orte_schizo_base_finalize(void);