smpd_launch_process.c

fortran并行计算包

    hStdin = GetStdHandle(STD_INPUT_HANDLE);
    hStdout = GetStdHandle(STD_OUTPUT_HANDLE);
    hStderr = GetStdHandle(STD_ERROR_HANDLE);
    if (hStdin == INVALID_HANDLE_VALUE || hStdout == INVALID_HANDLE_VALUE  || hStderr == INVALID_HANDLE_VALUE)
    {
	nError = GetLastError(); /* This will only be correct if stderr failed */
	ReleaseMutex(smpd_process.hLaunchProcessMutex);
	smpd_translate_win_error(nError, err_msg, MAX_ERROR_LENGTH, NULL);
	smpd_err_printf("GetStdHandle failed, error %d, %s\n", nError, err_msg);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;;
    }

    /* Create sockets for stdin, stdout, and stderr */
    nError = smpd_make_socket_loop_choose(&hSockStdinR, SMPD_FALSE, &hSockStdinW, SMPD_TRUE);
    if (nError)
    {
	smpd_err_printf("smpd_make_socket_loop failed, error %d\n", nError);
	goto CLEANUP;
    }
    nError = smpd_make_socket_loop_choose(&hSockStdoutR, SMPD_TRUE, &hSockStdoutW, SMPD_FALSE);
    if (nError)
    {
	smpd_err_printf("smpd_make_socket_loop failed, error %d\n", nError);
	goto CLEANUP;
    }
    nError = smpd_make_socket_loop_choose(&hSockStderrR, SMPD_TRUE, &hSockStderrW, SMPD_FALSE);
    if (nError)
    {
	smpd_err_printf("smpd_make_socket_loop failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (process->pmi != NULL)
    {
	if (smpd_process.use_inherited_handles)
	{
	    nError = smpd_make_socket_loop(&hSockPmiR, &hSockPmiW);
	    if (nError)
	    {
		smpd_err_printf("smpd_make_socket_loop failed, error %d\n", nError);
		goto CLEANUP;
	    }
	}
	else
	{
	    nError = MPIDU_Sock_listen(set, NULL, &listener_port, &sock_pmi_listener); 
	    if (nError != MPI_SUCCESS)
	    {
		smpd_err_printf("MPIDU_Sock_listen failed,\nsock error: %s\n", get_sock_error_string(nError));
		goto CLEANUP;
	    }
	    smpd_dbg_printf("pmi listening on port %d\n", listener_port);

	    nError = smpd_create_context(SMPD_CONTEXT_PMI_LISTENER, set, sock_pmi_listener, -1, &listener_context);
	    if (nError != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to create a context for the pmi listener.\n");
		goto CLEANUP;
	    }
	    nError = MPIDU_Sock_set_user_ptr(sock_pmi_listener, listener_context);
	    if (nError != MPI_SUCCESS)
	    {
		smpd_err_printf("MPIDU_Sock_set_user_ptr failed,\nsock error: %s\n", get_sock_error_string(nError));
		goto CLEANUP;
	    }
	    listener_context->state = SMPD_PMI_LISTENING;
	    /* Adding process rank since the interface for MPIDU_Sock_get_host_description changed */
	    nError = MPIDU_Sock_get_host_description(process->rank, host_description, 256);
	    if (nError != MPI_SUCCESS)
	    {
		smpd_err_printf("MPIDU_Sock_get_host_description failed,\nsock error: %s\n", get_sock_error_string(nError));
		goto CLEANUP;
	    }
	    /* save the listener sock in the slot reserved for the client sock so we can match the listener
	    to the process structure when the client sock is accepted */
	    process->pmi->sock = sock_pmi_listener;
	}
    }

    saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
    saAttr.lpSecurityDescriptor = NULL;
    saAttr.bInheritHandle = TRUE;

    /* Create the pipes for stdout, stderr */
    if (!CreatePipe(&hPipeStdinR, &hPipeStdinW, &saAttr, 0))
    {
	smpd_err_printf("CreatePipe(stdin) failed, error %d\n", GetLastError());
	goto CLEANUP;
    }
    if (!CreatePipe(&hPipeStdoutR, &hPipeStdoutW, &saAttr, 0))
    {
	smpd_err_printf("CreatePipe(stdout) failed, error %d\n", GetLastError());
	goto CLEANUP;
    }
    if (!CreatePipe(&hPipeStderrR, &hPipeStderrW, &saAttr, 0))
    {
	smpd_err_printf("CreatePipe(stderr) failed, error %d\n", GetLastError());
	goto CLEANUP;
    }

    /* Make the ends of the pipes that this process will use not inheritable */
    /*
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdinW, GetCurrentProcess(), &hIn, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    */
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hPipeStdinW, GetCurrentProcess(), &hIn, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    /*
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdoutR, GetCurrentProcess(), &hOut, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStderrR, GetCurrentProcess(), &hErr, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    */
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hPipeStdoutR, GetCurrentProcess(), &hOut, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hPipeStderrR, GetCurrentProcess(), &hErr, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (process->pmi != NULL && smpd_process.use_inherited_handles)
    {
	if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockPmiR, GetCurrentProcess(), (LPHANDLE)&hSockPmiR, 
	    0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
	{
	    nError = GetLastError();
	    smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	    goto CLEANUP;
	}
    }

    /* prevent the socket loops from being inherited */
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdinR, GetCurrentProcess(), (LPHANDLE)&hSockStdinR, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdoutR, GetCurrentProcess(), (LPHANDLE)&hSockStdoutR, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStderrR, GetCurrentProcess(), (LPHANDLE)&hSockStderrR, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdinW, GetCurrentProcess(), (LPHANDLE)&hSockStdinW, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStdoutW, GetCurrentProcess(), (LPHANDLE)&hSockStdoutW, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    if (!DuplicateHandle(GetCurrentProcess(), (HANDLE)hSockStderrW, GetCurrentProcess(), (LPHANDLE)&hSockStderrW, 
	0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
    {
	nError = GetLastError();
	smpd_err_printf("DuplicateHandle failed, error %d\n", nError);
	goto CLEANUP;
    }

    /* make the ends used by the spawned process blocking */
    /*
    blocking_flag = 0;
    ioctlsocket(hSockStdinR, FIONBIO, &blocking_flag);
    */
    /*
    blocking_flag = 0;
    ioctlsocket(hSockStdoutW, FIONBIO, &blocking_flag);
    blocking_flag = 0;
    ioctlsocket(hSockStderrW, FIONBIO, &blocking_flag);
    */

    /* Set stdin, stdout, and stderr to the ends of the pipe the created process will use */
    /*
    if (!SetStdHandle(STD_INPUT_HANDLE, (HANDLE)hSockStdinR))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    */
    if (!SetStdHandle(STD_INPUT_HANDLE, (HANDLE)hPipeStdinR))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto CLEANUP;
    }
    /*
    if (!SetStdHandle(STD_OUTPUT_HANDLE, (HANDLE)hSockStdoutW))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto RESTORE_CLEANUP;
    }
    if (!SetStdHandle(STD_ERROR_HANDLE, (HANDLE)hSockStderrW))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto RESTORE_CLEANUP;
    }
    */
    if (!SetStdHandle(STD_OUTPUT_HANDLE, (HANDLE)hPipeStdoutW))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto RESTORE_CLEANUP;
    }
    if (!SetStdHandle(STD_ERROR_HANDLE, (HANDLE)hPipeStderrW))
    {
	nError = GetLastError();
	smpd_err_printf("SetStdHandle failed, error %d\n", nError);
	goto RESTORE_CLEANUP;
    }

    /* Create the process */
    memset(&saInfo, 0, sizeof(STARTUPINFO));
    saInfo.cb = sizeof(STARTUPINFO);
    /*saInfo.hStdInput = (HANDLE)hSockStdinR;*/
    saInfo.hStdInput = (HANDLE)hPipeStdinR;
    /*
    saInfo.hStdOutput = (HANDLE)hSockStdoutW;
    saInfo.hStdError = (HANDLE)hSockStderrW;
    */
    saInfo.hStdOutput = (HANDLE)hPipeStdoutW;
    saInfo.hStdError = (HANDLE)hPipeStderrW;
    saInfo.dwFlags = STARTF_USESTDHANDLES;

    SetEnvironmentVariables(process->env);
    if (process->pmi != NULL)
    {
	sprintf(str, "%d", process->rank);
	smpd_dbg_printf("env: PMI_RANK=%s\n", str);
	SetEnvironmentVariable("PMI_RANK", str);
	sprintf(str, "%d", process->nproc);
	smpd_dbg_printf("env: PMI_SIZE=%s\n", str);
	SetEnvironmentVariable("PMI_SIZE", str);
	sprintf(str, "%s", process->kvs_name);
	smpd_dbg_printf("env: PMI_KVS=%s\n", str);
	SetEnvironmentVariable("PMI_KVS", str);
	sprintf(str, "%s", process->domain_name);
	smpd_dbg_printf("env: PMI_DOMAIN=%s\n", str);
	SetEnvironmentVariable("PMI_DOMAIN", str);
	if (smpd_process.use_inherited_handles)
	{
	    smpd_encode_handle(sock_str, (HANDLE)hSockPmiW);
	    sprintf(str, "%s", sock_str);
	    smpd_dbg_printf("env: PMI_SMPD_FD=%s\n", str);
	    SetEnvironmentVariable("PMI_SMPD_FD", str);
	}
	else
	{
	    smpd_dbg_printf("env: PMI_HOST=%s\n", host_description);
	    SetEnvironmentVariable("PMI_HOST", host_description);
	    sprintf(str, "%d", listener_port);
	    smpd_dbg_printf("env: PMI_PORT=%s\n", str);
	    SetEnvironmentVariable("PMI_PORT", str);
	}
	sprintf(str, "%d", smpd_process.id);
	smpd_dbg_printf("env: PMI_SMPD_ID=%s\n", str);
	SetEnvironmentVariable("PMI_SMPD_ID", str);
	sprintf(str, "%d", process->id);
	smpd_dbg_printf("env: PMI_SMPD_KEY=%s\n", str);
	SetEnvironmentVariable("PMI_SMPD_KEY", str);
	if (process->clique[0] != '\0')
	{
	    sprintf(str, "%s", process->clique);
	    smpd_dbg_printf("env: PMI_CLIQUE=%s\n", str);
	    SetEnvironmentVariable("PMI_CLIQUE", str);
	}
	sprintf(str, "%d", process->spawned);
	smpd_dbg_printf("env: PMI_SPAWN=%s\n", str);
	SetEnvironmentVariable("PMI_SPAWN", str);
	sprintf(str, "%d", process->appnum);
	smpd_dbg_printf("env: PMI_APPNUM=%s\n", str);
	SetEnvironmentVariable("PMI_APPNUM", str);
    }
    pEnv = GetEnvironmentStrings();

    GetCurrentDirectory(MAX_PATH, tSavedPath);
    SetCurrentDirectory(process->dir);

    launch_flag = 
	CREATE_SUSPENDED | /*CREATE_NEW_CONSOLE*/ /*DETACHED_PROCESS*/ CREATE_NO_WINDOW | priorityClass;
    if (dbg)
	launch_flag = launch_flag | DEBUG_PROCESS;

    smpd_dbg_printf("CreateProcess(%s)\n", actual_exe);
    psInfo.hProcess = INVALID_HANDLE_VALUE;
    if (CreateProcess(
	NULL,
	actual_exe,
	NULL, NULL, TRUE,
	launch_flag,
	pEnv,
	NULL,
	&saInfo, &psInfo))
    {
	SetThreadPriority(psInfo.hThread, priority);
	process->pid = psInfo.dwProcessId;
    }
    else
    {
	nError = GetLastError();
	smpd_err_printf("CreateProcess('%s') failed, error %d\n", process->exe, nError);
	smpd_translate_win_error(nError, process->err_msg, SMPD_MAX_ERROR_LEN, "CreateProcess(%s) on '%s' failed, error %d - ",
	    process->exe, smpd_process.host, nError);
	psInfo.hProcess = INVALID_HANDLE_VALUE;
	bSuccess = FALSE;
    }

    FreeEnvironmentStrings((TCHAR*)pEnv);
    SetCurrentDirectory(tSavedPath);
    RemoveEnvironmentVariables(process->env);
    if (process->pmi != NULL)
    {
	SetEnvironmentVariable("PMI_RANK", NULL);
	SetEnvironmentVariable("PMI_SIZE", NULL);
	SetEnvironmentVariable("PMI_KVS", NULL);
	SetEnvironmentVariable("PMI_DOMAIN", NULL);
	if (smpd_process.use_inherited_handles)
	{
	    SetEnvironmentVariable("PMI_SMPD_FD", NULL);
	}
	else
	{
	    SetEnvironmentVariable("PMI_HOST", NULL);
	    SetEnvironmentVariable("PMI_PORT", NULL);
	}
	SetEnvironmentVariable("PMI_SMPD_ID", NULL);
	SetEnvironmentVariable("PMI_SMPD_KEY", NULL);
	SetEnvironmentVariable("PMI_SPAWN", NULL);
    }

    if (bSuccess)
    {
	/* make sock structures out of the sockets */
	/*
	nError = MPIDU_Sock_native_to_sock(set, hIn, NULL, &sock_in);
	if (nError != MPI_SUCCESS)
	{
	    smpd_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(nError));
	}
	*/
	/*printf("native sock for stdin\n");fflush(stdout);*/
	nError = MPIDU_Sock_native_to_sock(set, (MPIDU_SOCK_NATIVE_FD)hSockStdinW, NULL, &sock_in);
	if (nError != MPI_SUCCESS)
	{
	    smpd_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(nError));
	}
	/*printf("native sock for stdout\n");fflush(stdout);*/
	nError = MPIDU_Sock_native_to_sock(set, (MPIDU_SOCK_NATIVE_FD)hSockStdoutR, NULL, &sock_out);
	if (nError != MPI_SUCCESS)
	{
	    smpd_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(nError));
	}
	/*printf("native sock for stderr\n");fflush(stdout);*/
	nError = MPIDU_Sock_native_to_sock(set, (MPIDU_SOCK_NATIVE_FD)hSockStderrR, NULL, &sock_err);
	if (nError != MPI_SUCCESS)
	{
	    smpd_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(nError));
	}
	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	{
	    /*printf("native sock for pmi\n");fflush(stdout);*/
	    nError = MPIDU_Sock_native_to_sock(set, (MPIDU_SOCK_NATIVE_FD)hSockPmiR, NULL, &sock_pmi);
	    if (nError != MPI_SUCCESS)
	    {
		smpd_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(nError));
	    }
	}

	process->in->sock = sock_in;
	process->out->sock = sock_out;
	process->err->sock = sock_err;
	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	    process->pmi->sock = sock_pmi;
	process->pid = process->in->id = process->out->id = process->err->id = psInfo.dwProcessId;
	MPIDU_Sock_set_user_ptr(sock_in, process->in);
	MPIDU_Sock_set_user_ptr(sock_out, process->out);
	MPIDU_Sock_set_user_ptr(sock_err, process->err);
	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	    MPIDU_Sock_set_user_ptr(sock_pmi, process->pmi);
    }
    else
    {
	/* close all the sockets and handles allocated in this function */
	/*CloseHandle(hIn);*/
	CloseHandle((HANDLE)hSockStdinW);
	CloseHandle((HANDLE)hSockStdoutR);
	CloseHandle((HANDLE)hSockStderrR);
	if (process->pmi != NULL && smpd_process.use_inherited_handles)
	    CloseHandle((HANDLE)hSockPmiR);
    }

RESTORE_CLEANUP:
    /* Restore stdin, stdout, stderr */
    SetStdHandle(STD_INPUT_HANDLE, hStdin);
    SetStdHandle(STD_OUTPUT_HANDLE, hStdout);
    SetStdHandle(STD_ERROR_HANDLE, hStderr);

CLEANUP:
    ReleaseMutex(smpd_process.hLaunchProcessMutex);
    /*CloseHandle((HANDLE)hSockStdinR);*/
    CloseHandle((HANDLE)hPipeStdinR);
    /*
    CloseHandle((HANDLE)hSockStdoutW);
    CloseHandle((HANDLE)hSockStderrW);
    */
    CloseHandle((HANDLE)hPipeStdoutW);
    CloseHandle((HANDLE)hPipeStderrW);
    if (process->pmi != NULL && smpd_process.use_inherited_handles)
	CloseHandle((HANDLE)hSockPmiW);

    if (psInfo.hProcess != INVALID_HANDLE_VALUE)
    {
	HANDLE hThread;
	smpd_piothread_arg_t *arg_ptr;
	smpd_pinthread_arg_t *in_arg_ptr;

	in_arg_ptr = (smpd_pinthread_arg_t*)malloc(sizeof(smpd_pinthread_arg_t));
	in_arg_ptr->hIn = hSockStdinR;
	in_arg_ptr->hOut = hPipeStdinW;
	in_arg_ptr->pid = psInfo.dwProcessId;
	hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)smpd_pinthread, in_arg_ptr, 0, NULL);
	CloseHandle(hThread);
	arg_ptr = (smpd_piothread_arg_t*)malloc(sizeof(smpd_piothread_arg_t));