/* * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana * University Research and Technology * Corporation. All rights reserved. * Copyright (c) 2004-2008 The University of Tennessee and The University * of Tennessee Research Foundation. All rights * reserved. * Copyright (c) 2004-2006 High Performance Computing Center Stuttgart, * University of Stuttgart. All rights reserved. * Copyright (c) 2004-2005 The Regents of the University of California. * All rights reserved. * Copyright (c) 2006 Cisco Systems, Inc. All rights reserved. * Copyright (c) 2007 Los Alamos National Security, LLC. All rights * reserved. * $COPYRIGHT$ * * Additional copyrights may follow * * $HEADER$ * * These symbols are in a file by themselves to provide nice linker * semantics. Since linkers generally pull in symbols by object * files, keeping these symbols as the only symbols in this file * prevents utility programs such as "ompi_info" from having to import * entire components just to query their version and parameters. */ #include "orte_config.h" #include "orte/constants.h" #include #ifdef HAVE_UNISTD_H #include #endif #include #include #ifdef HAVE_SYS_SELECT_H #include #endif #ifdef HAVE_SYS_TIME_H #include #endif #ifdef HAVE_SYS_TYPES_H #include #endif #ifdef HAVE_SYS_STAT_H #include #endif #ifdef HAVE_SYS_WAIT_H #include #endif #include #include #ifdef HAVE_PWD_H #include #endif #include "opal/mca/installdirs/installdirs.h" #include "opal/mca/base/mca_base_param.h" #include "opal/util/if.h" #include "opal/util/os_path.h" #include "opal/util/path.h" #include "opal/event/event.h" #include "opal/util/argv.h" #include "opal/util/opal_environ.h" #include "orte/util/show_help.h" #include "opal/util/trace.h" #include "opal/util/basename.h" #include "opal/util/opal_environ.h" #include "orte/util/session_dir.h" #include "orte/util/name_fns.h" #include "orte/runtime/orte_wait.h" #include "opal/dss/dss.h" #include "orte/mca/rml/rml.h" #include "orte/mca/errmgr/errmgr.h" #include "orte/mca/ras/ras_types.h" #include "orte/mca/rmaps/rmaps.h" #include "orte/mca/plm/plm.h" #include "orte/mca/plm/base/base.h" #include "orte/mca/plm/base/plm_private.h" #include "orte/mca/plm/process/plm_process.h" #define rindex(a,b) strrchr((a),(b)) //daniel #if OMPI_HAVE_POSIX_THREADS && OMPI_THREADS_HAVE_DIFFERENT_PIDS && OMPI_ENABLE_PROGRESS_THREADS static int orte_plm_process_launch_threaded(orte_jobid_t jobid); #endif /* * Interface */ static int orte_plm_process_init(void); static int orte_plm_process_set_hnp_name(void); static int orte_plm_process_launch(orte_job_t*); static int orte_plm_process_terminate_job(orte_jobid_t); static int orte_plm_process_terminate_orteds(void); static int orte_plm_process_signal_job(orte_jobid_t, int32_t); orte_plm_base_module_t orte_plm_process_module = { orte_plm_process_init, orte_plm_base_set_hnp_name, #if OMPI_HAVE_POSIX_THREADS && OMPI_THREADS_HAVE_DIFFERENT_PIDS && OMPI_ENABLE_PROGRESS_THREADS orte_plm_process_launch_threaded, #else orte_plm_process_launch, #endif NULL, orte_plm_process_terminate_job, orte_plm_process_terminate_orteds, orte_plm_process_signal_job, orte_plm_process_finalize }; static void set_handler_default(int sig); enum { ORTE_PLM_RSH_SHELL_BASH = 0, ORTE_PLM_RSH_SHELL_ZSH, ORTE_PLM_RSH_SHELL_TCSH, ORTE_PLM_RSH_SHELL_CSH, ORTE_PLM_RSH_SHELL_KSH, ORTE_PLM_RSH_SHELL_SH, ORTE_PLM_RSH_SHELL_UNKNOWN }; typedef int orte_plm_process_shell; static const char * orte_plm_process_shell_name[] = { "bash", "zsh", "tcsh", /* tcsh has to be first otherwise strstr finds csh */ "csh", "ksh", "sh", "unknown" }; /* local global storage of timing variables */ static struct timeval joblaunchstart, joblaunchstop; /* global storage of active jobid being launched */ static orte_jobid_t active_job = ORTE_JOBID_INVALID; /** * Init the module */ int orte_plm_process_init(void) { int rc; if (ORTE_SUCCESS != (rc = orte_plm_base_comm_start())) { ORTE_ERROR_LOG(rc); } return rc; } /** * Check the Shell variable on the specified node */ static int orte_plm_process_probe(orte_node_t * node, orte_plm_process_shell * shell) { char ** argv; int rc, nfds; int fd[2]; pid_t pid; /* HANDLE myPipeFd[2]; SECURITY_ATTRIBUTES securityAttr; STARTUPINFO startupInfo; PROCESS_INFORMATION processInfo; */ fd_set readset; fd_set errset; char outbuf[4096]; OPAL_OUTPUT_VERBOSE((1, orte_plm_globals.output, "%s plm:process: going to check SHELL variable on node %s", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), node->name)); *shell = ORTE_PLM_RSH_SHELL_UNKNOWN; /* * Build argv array */ pid = _spawnve( _P_DETACH, argv[0], argv, NULL); //daniel #if 0 securityAttr.nLength = sizeof(SECURITY_ATTRIBUTES); // Size of struct securityAttr.lpSecurityDescriptor = NULL; // Default descriptor securityAttr.bInheritHandle = TRUE; // Inheritable // Create the pipe if (CreatePipe(&myPipeFd[0], &myPipeFd[1], &securityAttr, 0)) { // Create duplicate of write end so that original if (!DuplicateHandle( GetCurrentProcess(), myPipeFd[0], // Original handle GetCurrentProcess(), NULL, // don't create new handle 0, FALSE, // Not inheritable DUPLICATE_SAME_ACCESS) ) { CloseHandle(myPipeFd[0]); CloseHandle(myPipeFd[1]); opal_output(0, "plm:process: DuplicateHandle failed with errno=%d\n", errno); return ORTE_ERR_IN_ERRNO; } ZeroMemory( &startupInfo, sizeof(startupInfo) ); startupInfo.cb = sizeof(startupInfo); ZeroMemory( &processInfo, sizeof(processInfo) ); // Now populate startup info for CreateProcess GetStartupInfo(&startupInfo); startupInfo.dwFlags = STARTF_USESTDHANDLES; startupInfo.hStdInput = myPipeFd[0]; startupInfo.hStdOutput = GetStdHandle(STD_OUTPUT_HANDLE); startupInfo.hStdError = GetStdHandle(STD_ERROR_HANDLE); // Start the child process. if( !CreateProcess( argv[0], //module name NULL, NULL, //(LPSTR)(const char *) argv, NULL, // Process handle not inheritable NULL, // Thread handle not inheritable TRUE, // Set handle inheritance to TRUE; // each inheritable handle in the calling process is inherited by the new process 0, // No creation flags NULL, // Use parent's environment block NULL, // Use parent's starting directory &startupInfo, // Pointer to STARTUPINFO structure &processInfo ) // Pointer to PROCESS_INFORMATION structure ) { CloseHandle(myPipeFd[1]); opal_output(0, "plm:process: CreateProcess failed with errno=%d\n", errno); //, GetLastError() ); return ORTE_ERR_IN_ERRNO; } } #endif /* if ((pid = fork()) < 0) { opal_output(0, "plm:process: fork failed with errno=%d\n", errno); return ORTE_ERR_IN_ERRNO; } else if (pid == 0) { // child //processInfo.hProcess if (dup2(fd[1], 1) < 0) { opal_output(0, "plm:process: dup2 failed with errno=%d\n", errno); return ORTE_ERR_IN_ERRNO; } execvp(argv[0], argv); exit(errno); } if (close(fd[1])) { opal_output(0, "plm:process: close failed with errno=%d\n", errno); return ORTE_ERR_IN_ERRNO; } */ /* Monitor stdout */ FD_ZERO(&readset); nfds = fd[0]+1; memset (outbuf, 0, sizeof (outbuf)); rc = ORTE_SUCCESS;; while (ORTE_SUCCESS == rc) { int err; FD_SET (fd[0], &readset); errset = readset; err = select(nfds, &readset, NULL, &errset, NULL); if (err == -1) { if (errno == EINTR) continue; else { rc = ORTE_ERR_IN_ERRNO; break; } } if (FD_ISSET(fd[0], &errset) != 0) rc = ORTE_ERR_FATAL; /* In case we have something valid to read on stdin */ if (FD_ISSET(fd[0], &readset) != 0) { ssize_t ret = 1; char temp[4096]; char * ptr = outbuf; ssize_t outbufsize = sizeof(outbuf); memset (temp, 0, sizeof(temp)); while (ret != 0) { ret = read (fd[0], temp, 256); if (ret < 0) { if (errno == EINTR) continue; else { rc = ORTE_ERR_IN_ERRNO; break; } } else { if (outbufsize > 0) { memcpy (ptr, temp, (ret > outbufsize) ? outbufsize : ret); outbufsize -= ret; ptr += ret; if (outbufsize > 0) *ptr = '\0'; } } } /* After reading complete string (aka read returns 0), we just break */ break; } } /* Search for the substring of known shell-names */ /* for (i = 0; i < (int)(sizeof (orte_plm_process_shell_name)/ sizeof(orte_plm_process_shell_name[0])); i++) { char *sh_name = NULL; sh_name = rindex(outbuf, '/'); if ( sh_name != NULL ) { sh_name++; /* skip '/' */ /* We cannot use "echo -n $SHELL" because -n is not portable. Therefore * we have to remove the "\n" */ /* if ( sh_name[strlen(sh_name)-1] == '\n' ) { sh_name[strlen(sh_name)-1] = '\0'; } if ( 0 == strcmp(sh_name, orte_plm_process_shell_name[i]) ) { *shell = i; break; } } } */ OPAL_OUTPUT_VERBOSE((1, orte_plm_globals.output, "%s plm:process: node:%s has SHELL: %s", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), node->name, orte_plm_process_shell_name[*shell])); return rc; } /** * Fill the exec_path variable with the directory to the orted */ static int orte_plm_process_fill_exec_path( char ** exec_path ) { struct stat buf; asprintf(exec_path, "%s/orted", opal_install_dirs.bindir); if (0 != stat(*exec_path, &buf)) { char *path = getenv("PATH"); if (NULL == path) { path = "PATH is empty!"; } orte_show_help("help-plm-process.txt", "no-local-orted", true, path, opal_install_dirs.bindir); return ORTE_ERR_NOT_FOUND; } return ORTE_SUCCESS; } /** * Callback on daemon exit. */ static void orte_plm_process_wait_daemon(pid_t pid, int status, void* cbdata) { unsigned long deltat; if (! WIFEXITED(status) || ! WEXITSTATUS(status) == 0) { /* tell the user something went wrong */ opal_output(0, "ERROR: A daemon failed to start as expected."); opal_output(0, "ERROR: There may be more information available from"); opal_output(0, "ERROR: the remote shell (see above)."); if (WIFEXITED(status)) { opal_output(0, "ERROR: The daemon exited unexpectedly with status %d.", WEXITSTATUS(status)); } else if (WIFSIGNALED(status)) { #ifdef WCOREDUMP if (WCOREDUMP(status)) { opal_output(0, "The daemon received a signal %d (with core).", WTERMSIG(status)); } else { opal_output(0, "The daemon received a signal %d.", WTERMSIG(status)); } #else opal_output(0, "The daemon received a signal %d.", WTERMSIG(status)); #endif /* WCOREDUMP */ } else { opal_output(0, "No extra status information is available: %d.", status); } /* report that the daemon has failed so we break out of the daemon * callback receive and can exit */ orte_plm_base_launch_failed(active_job, pid, status, ORTE_JOB_STATE_FAILED_TO_START); } /* if abnormal exit */ /* release any waiting threads */ OPAL_THREAD_LOCK(&mca_plm_process_component.lock); if (mca_plm_process_component.num_children-- >= mca_plm_process_component.num_concurrent || mca_plm_process_component.num_children == 0) { opal_condition_signal(&mca_plm_process_component.cond); } if (mca_plm_process_component.timing && mca_plm_process_component.num_children == 0) { if (0 != gettimeofday(&joblaunchstop, NULL)) { opal_output(0, "plm_process: could not obtain job launch stop time"); } else { deltat = (joblaunchstop.tv_sec - joblaunchstart.tv_sec)*1000000 + (joblaunchstop.tv_usec - joblaunchstart.tv_usec); opal_output(0, "plm_process: total time to launch job is %lu usec", deltat); } } OPAL_THREAD_UNLOCK(&mca_plm_process_component.lock); } /** * Launch a daemon (bootproxy) on each node. The daemon will be responsible * for launching the application. */ /* When working in this function, ALWAYS jump to "cleanup" if * you encounter an error so that orterun will be woken up and * the job can cleanly terminate */ int orte_plm_process_launch(orte_job_t *jdata) { orte_job_map_t *map = NULL; int proc_vpid_index; int local_exec_index; char *vpid_string = NULL; char *param; char **argv = NULL; char *prefix_dir; int argc = 0; int rc; char *lib_base = NULL, *bin_base = NULL; bool failed_launch = true; orte_app_context_t **apps; orte_node_t **nodes; orte_std_cntr_t nnode; orte_job_state_t job_state = ORTE_JOB_NEVER_LAUNCHED; if (mca_plm_process_component.timing) { if (0 != gettimeofday(&joblaunchstart, NULL)) { opal_output(0, "plm_process: could not obtain start time"); joblaunchstart.tv_sec = 0; joblaunchstart.tv_usec = 0; } } /* create a jobid for this job */ if (ORTE_SUCCESS != (rc = orte_plm_base_create_jobid(&jdata->jobid))) { ORTE_ERROR_LOG(rc); return rc; } OPAL_OUTPUT_VERBOSE((1, orte_plm_globals.output, "%s plm:process: launching job %s", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), ORTE_JOBID_PRINT(jdata->jobid))); /* setup the job */ if (ORTE_SUCCESS != (rc = orte_plm_base_setup_job(jdata))) { ORTE_ERROR_LOG(rc); goto cleanup; } /* set the active jobid */ active_job = jdata->jobid; /* Get the map for this job */ if (NULL == (map = orte_rmaps.get_job_map(active_job))) { ORTE_ERROR_LOG(ORTE_ERR_NOT_FOUND); rc = ORTE_ERR_NOT_FOUND; goto cleanup; } apps = (orte_app_context_t**)jdata->apps->addr; nodes = (orte_node_t**)map->nodes->addr; if (0 == map->num_new_daemons) { /* have all the daemons we need - launch app */ OPAL_OUTPUT_VERBOSE((1, orte_plm_globals.output, "%s plm:process: no new daemons to launch", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME))); goto launch_apps; } if (orte_debug_daemons_flag && mca_plm_process_component.num_concurrent < map->num_new_daemons) { /** * If we are in '--debug-daemons' we keep the ssh connection * alive for the span of the run. If we use this option * AND we launch on more than "num_concurrent" machines * then we will deadlock. No connections are terminated * until the job is complete, no job is started * since all the orteds are waiting for all the others * to come online, and the others ore not launched because * we are waiting on those that have started to terminate * their ssh tunnels. :( * As we cannot run in this situation, pretty print the error * and return an error code. */ orte_show_help("help-plm-process.txt", "deadlock-params", true, mca_plm_process_component.num_concurrent, map->num_new_daemons); rc = ORTE_ERR_FATAL; goto cleanup; } /* * After a discussion between Ralph & Jeff, we concluded that we * really are handling the prefix dir option incorrectly. It currently * is associated with an app_context, yet it really refers to the * location where OpenRTE/Open MPI is installed on a NODE. Fixing * this right now would involve significant change to orterun as well * as elsewhere, so we will intentionally leave this incorrect at this * point. The error, however, is identical to that seen in all prior * releases of OpenRTE/Open MPI, so our behavior is no worse than before. * * A note to fix this, along with ideas on how to do so, has been filed * on the project's Trac system under "feature enhancement". * * For now, default to the prefix_dir provided in the first app_context. * Since there always MUST be at least one app_context, we are safe in * doing this. */ prefix_dir = apps[0]->prefix_dir; /* * Build argv array */ opal_argv_append(&argc, &argv, "