smpd_launch_process.c

fortran并行计算包

	arg_ptr->hIn = hOut;
	arg_ptr->hOut = hSockStdoutW;
	arg_ptr->pid = psInfo.dwProcessId;
	hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)smpd_piothread, arg_ptr, 0, NULL);
	CloseHandle(hThread);
	arg_ptr = (smpd_piothread_arg_t*)malloc(sizeof(smpd_piothread_arg_t));
	arg_ptr->hIn = hErr;
	arg_ptr->hOut = hSockStderrW;
	arg_ptr->pid = psInfo.dwProcessId;
	hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)smpd_piothread, arg_ptr, 0, NULL);
	CloseHandle(hThread);

	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	    process->context_refcount = 3;
	else
	    process->context_refcount = 2;
	process->out->read_state = SMPD_READING_STDOUT;
	result = MPIDU_Sock_post_read(sock_out, process->out->read_cmd.cmd, 1, 1, NULL);
	if (result != MPI_SUCCESS)
	{
	    smpd_err_printf("posting first read from stdout context failed, sock error: %s\n",
		get_sock_error_string(result));
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	process->err->read_state = SMPD_READING_STDERR;
	result = MPIDU_Sock_post_read(sock_err, process->err->read_cmd.cmd, 1, 1, NULL);
	if (result != MPI_SUCCESS)
	{
	    smpd_err_printf("posting first read from stderr context failed, sock error: %s\n",
		get_sock_error_string(result));
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	{
	    result = smpd_post_read_command(process->pmi);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to post a read of the first command on the pmi control channel.\n");
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	}
	process->wait.hProcess = process->in->wait.hProcess = process->out->wait.hProcess = process->err->wait.hProcess = psInfo.hProcess;
	process->wait.hThread = process->in->wait.hThread = process->out->wait.hThread = process->err->wait.hThread = psInfo.hThread;
	smpd_process_to_registry(process, actual_exe);
	ResumeThread(psInfo.hThread);
	smpd_exit_fn(FCNAME);
	return SMPD_SUCCESS;
    }

    smpd_exit_fn(FCNAME);
    return SMPD_FAIL;
}

#undef FCNAME
#define FCNAME "smpd_parse_account_domain"
void smpd_parse_account_domain(const char *domain_account, char *account, char *domain)
{
    const char *pCh;
    char *pCh2;

    smpd_enter_fn(FCNAME);

    if ((strchr(domain_account, '\\') == NULL) && (strchr(domain_account, '@') != NULL))
    {
	pCh = domain_account;
	pCh2 = account;
	while ((*pCh != '@') && (*pCh != '\0'))
	{
	    *pCh2 = *pCh;
	    pCh++;
	    pCh2++;
	}
	*pCh2 = '\0';
	if (*pCh == '@')
	{
	    pCh++;
	    strcpy(domain, pCh);
	}
	else
	{
	    domain[0] = '\0';
	}
    }
    else
    {
	pCh = domain_account;
	pCh2 = domain;
	while ((*pCh != '\\') && (*pCh != '\0'))
	{
	    *pCh2 = *pCh;
	    pCh++;
	    pCh2++;
	}
	if (*pCh == '\\')
	{
	    pCh++;
	    strcpy(account, pCh);
	    *pCh2 = '\0';
	}
	else
	{
	    strcpy(account, domain_account);
	    domain[0] = '\0';
	}
    }

    smpd_exit_fn(FCNAME);
}

#else

/* Unix code */

/*
static void set_environment_variables(char *bEnv)
{
    char name[1024]="", value[8192]="";
    char *pChar;
    
    pChar = name;
    while (*bEnv != '\0')
    {
	if (*bEnv == '=')
	{
	    *pChar = '\0';
	    pChar = value;
	}
	else
	{
	    if (*bEnv == ';')
	    {
		*pChar = '\0';
		pChar = name;
		smpd_dbg_printf("env: %s=%s\n", name, value);
		setenv(name, value, 1);
	    }
	    else
	    {
		*pChar = *bEnv;
		pChar++;
	    }
	}
	bEnv++;
    }
    *pChar = '\0';
    if (name[0] != '\0')
    {
	smpd_dbg_printf("env: %s=%s\n", name, value);
	setenv(name, value, 1);
    }
}
*/

#ifdef HAVE_SETENV
#undef FCNAME
#define FCNAME "set_environment_variables"
static void set_environment_variables(char *bEnv)
{
    char name[1024], equals[3], value[8192];

    smpd_enter_fn(FCNAME);
    while (1)
    {
	name[0] = '\0';
	equals[0] = '\0';
	value[0] = '\0';
	if (MPIU_Str_get_string(&bEnv, name, 1024) != MPIU_STR_SUCCESS)
	    break;
	if (name[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, equals, 3) != MPIU_STR_SUCCESS)
	    break;
	if (equals[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, value, 8192) != MPIU_STR_SUCCESS)
	    break;
	setenv(name, value, 1);
    }
    smpd_exit_fn(FCNAME);
}
#else
#undef FCNAME
#define FCNAME "get_env_size"
static int get_env_size(char *bEnv, int *count)
{
    char name[1024], equals[3], value[8192];
    int size = 0;

    smpd_enter_fn(FCNAME);
    while (1)
    {
	name[0] = '\0';
	equals[0] = '\0';
	value[0] = '\0';
	if (MPIU_Str_get_string(&bEnv, name, 1024) != MPIU_STR_SUCCESS)
	    break;
	if (name[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, equals, 3) != MPIU_STR_SUCCESS)
	    break;
	if (equals[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, value, 8192) != MPIU_STR_SUCCESS)
	    break;
	*count = *count + 1;
	size = size + strlen(name) + strlen(value) + 2; /* length of 'name=value\0' */
    }
    smpd_exit_fn(FCNAME);
    return size;
}

#undef FCNAME
#define FCNAME "add_environment_variables"
static void add_environment_variables(char *str, char **vars, char *bEnv)
{
    char name[1024], equals[3], value[8192];
    int i = 0;

    smpd_enter_fn(FCNAME);
    while (1)
    {
	name[0] = '\0';
	equals[0] = '\0';
	value[0] = '\0';
	if (MPIU_Str_get_string(&bEnv, name, 1024) != MPIU_STR_SUCCESS)
	    break;
	if (name[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, equals, 3) != MPIU_STR_SUCCESS)
	    break;
	if (equals[0] == '\0')
	    break;
	if (MPIU_Str_get_string(&bEnv, value, 8192) != MPIU_STR_SUCCESS)
	    break;
	vars[i] = str;
	str += sprintf(str, "%s=%s", name, value) + 1;
	i++;
    }
    smpd_exit_fn(FCNAME);
}
#endif

#ifdef USE_PTHREAD_STDIN_REDIRECTION
static void child_exited(int signo)
{
    pid_t pid;
    int status = WNOHANG;
    smpd_process_t *iter;

    /*
     * For some reason on the Macs, stdout and stderr are not closed when the
     * process exits so we can't use the closing of the redirected stdout and
     * stderr sockets as indications that the process has exited.  So this
     * signal handler closes the stdout/err redirection socket on a SIGCHLD
     * signal to simulate that behavior.
     */
    if (signo == SIGCHLD)
    {
	smpd_dbg_printf("SIGCHLD received\n");
	for (;;)
	{
	    status = WNOHANG;
	    pid = wait(&status);
	    if (pid == -1)
		break;

	    smpd_dbg_printf("SIGCHILD pid = %d\n", pid);
	    iter = smpd_process.process_list;
	    {
		while (iter != NULL)
		{
		    if (iter->wait == pid)
		    {
			if (WIFEXITED(status))
			{
			    iter->exitcode =  WEXITSTATUS(status);
			}
			else
			{
			    iter->exitcode = -123;
			}
			if (iter->out != NULL)
			{
			    if (iter->out->read_state == SMPD_READING_STDOUT)
			    {
				smpd_dbg_printf("closing stdout redirection\n");
				iter->out->state = SMPD_CLOSING;
				MPIDU_Sock_post_close(iter->out->sock);
			    }
			    else
			    {
				smpd_err_printf("iter->out->read_state = %d\n", iter->out->read_state);
			    }
			}
			if (iter->err != NULL)
			{
			    if (iter->err->read_state == SMPD_READING_STDERR)
			    {
				smpd_dbg_printf("closing stderr redirection\n");
				iter->err->state = SMPD_CLOSING;
				MPIDU_Sock_post_close(iter->err->sock);
			    }
			    else
			    {
				smpd_err_printf("iter->err->read_state = %d\n", iter->err->read_state);
			    }
			}
			break;
		    }
		    iter = iter->next;
		}
	    }
	}
    }
    else
    {
	smpd_dbg_printf("unexpected signal %d received\n", signo);
    }
}
#endif

#undef FCNAME
#define FCNAME "smpd_launch_process"
int smpd_launch_process(smpd_process_t *process, int priorityClass, int priority, int dbg, MPIDU_Sock_set_t set)
{
    int result;
    int stdin_pipe_fds[2], stdout_pipe_fds[2];
    int  stderr_pipe_fds[2], pmi_pipe_fds[2];
    int pid;
    MPIDU_Sock_t sock_in, sock_out, sock_err, sock_pmi;
    char args[SMPD_MAX_EXE_LENGTH];
    char *argv[1024];
    char *token;
    int i;
    char str[1024];
    char *str_iter;
    int total, num_chars;
    char *actual_exe, exe_data[SMPD_MAX_EXE_LENGTH];
    char *temp_str;
    char temp_exe[SMPD_MAX_EXE_LENGTH];
    smpd_command_t *cmd_ptr;
    static char *pPutEnv = NULL;
    char *pLastEnv, *env_iter;
    int env_count, env_size;
    char **pEnvArray;

    smpd_enter_fn(FCNAME);

#ifdef USE_PTHREAD_STDIN_REDIRECTION
    {
	/* On the Macs we must use a signal handler to determine when a process
	 * has exited.  On all other systems we use the closing of stdout and
	 * stderr to determine that a process has exited.
	 */
	static int sighandler_setup = 0;
	if (!sighandler_setup)
	{
	    smpd_dbg_printf("setting child_exited signal handler\n");
	    smpd_signal(SIGCHLD, child_exited);
	    sighandler_setup = 1;
	}
    }
#endif

    /* resolve the executable name */
    if (process->path[0] != '\0')
    {
	temp_str = process->exe;
	result = MPIU_Str_get_string(&temp_str, temp_exe, SMPD_MAX_EXE_LENGTH);
	if (result != MPIU_STR_SUCCESS)
	{
	}
	smpd_dbg_printf("searching for '%s' in '%s'\n", temp_exe, process->path);
	if (smpd_search_path(process->path, temp_exe, SMPD_MAX_EXE_LENGTH, exe_data))
	{
	    smpd_dbg_printf("found: '%s'\n", exe_data);
	    if (strstr(exe_data, " "))
	    {
		smpd_err_printf("Currently unable to handle paths with spaces in them.\n");
	    }
	    if (temp_str != NULL)
	    {
		strncat(exe_data, " ", SMPD_MAX_EXE_LENGTH - strlen(exe_data));
		strncat(exe_data, temp_str, SMPD_MAX_EXE_LENGTH - strlen(exe_data));
		exe_data[SMPD_MAX_EXE_LENGTH-1] = '\0';
	    }
	    actual_exe = exe_data;
	}
	else
	{
	    actual_exe = process->exe;
	}
    }
    else
    {
	actual_exe = process->exe;
    }
    
    /* create argv from the command */
    i = 0;
    total = 0;
    /*str_iter = process->exe;*/
    str_iter = actual_exe;
    while (str_iter && i<1024)
    {
	result = MPIU_Str_get_string(&str_iter, &args[total],
				   SMPD_MAX_EXE_LENGTH - total);
	argv[i] = &args[total];
	i++;
	total += strlen(&args[total])+1; /* move over the null termination */
    }
    argv[i] = NULL;

    /* create pipes for redirecting I/O */
    /*
    pipe(stdin_pipe_fds);
    pipe(stdout_pipe_fds);
    pipe(stderr_pipe_fds);
    */
    socketpair(AF_UNIX, SOCK_STREAM, 0, stdin_pipe_fds);
    socketpair(AF_UNIX, SOCK_STREAM, 0, stdout_pipe_fds);
    socketpair(AF_UNIX, SOCK_STREAM, 0, stderr_pipe_fds);
    if (process->pmi != NULL)
    {
	socketpair(AF_UNIX, SOCK_STREAM, 0, pmi_pipe_fds);
    }

    pid = fork();
    if (pid < 0)
    {
	smpd_err_printf("fork failed - error %d.\n", errno);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    if (pid == 0)
    {
	/* child process */
	smpd_dbg_printf("client is alive and about to exec '%s'\n", argv[0]);

#ifdef HAVE_SETENV
	if (process->pmi != NULL)
	{
	    sprintf(str, "%d", process->rank);
	    smpd_dbg_printf("env: PMI_RANK=%s\n", str);
	    setenv("PMI_RANK", str, 1);
	    sprintf(str, "%d", process->nproc);
	    smpd_dbg_printf("env: PMI_SIZE=%s\n", str);
	    setenv("PMI_SIZE", str, 1);
	    sprintf(str, "%s", process->kvs_name);
	    smpd_dbg_printf("env: PMI_KVS=%s\n", str);
	    setenv("PMI_KVS", str, 1);
	    sprintf(str, "%s", process->domain_name);
	    smpd_dbg_printf("env: PMI_DOMAIN=%s\n", str);
	    setenv("PMI_DOMAIN", str, 1);
	    sprintf(str, "%d", pmi_pipe_fds[1]);
	    smpd_dbg_printf("env: PMI_SMPD_FD=%s\n", str);
	    setenv("PMI_SMPD_FD", str, 1);
	    sprintf(str, "%d", smpd_process.id);
	    smpd_dbg_printf("env: PMI_SMPD_ID=%s\n", str);
	    setenv("PMI_SMPD_ID", str, 1);
	    sprintf(str, "%d", process->id);
	    smpd_dbg_printf("env: PMI_SMPD_KEY=%s\n", str);
	    setenv("PMI_SMPD_KEY", str, 1);
	    sprintf(str, "%d", process->spawned);