smpd_launch_process.c

fortran并行计算包

    smpd_exit_fn(FCNAME);
}

#undef FCNAME
#define FCNAME "RemoveEnvironmentVariables"
static void RemoveEnvironmentVariables(char *bEnv)
{
    char name[SMPD_MAX_ENV_LENGTH], equals[3], value[SMPD_MAX_ENV_LENGTH];

    smpd_enter_fn(FCNAME);
    for (;;)
    {
	name[0] = '\0';
	equals[0] = '\0';
	value[0] = '\0';
	if (MPIU_Str_get_string(&bEnv, name, SMPD_MAX_ENV_LENGTH) != 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, SMPD_MAX_ENV_LENGTH) != MPIU_STR_SUCCESS)
	    break;
	/*smpd_dbg_printf("removing environment variable <%s>\n", name);*/
	SetEnvironmentVariable(name, NULL);
    }
    smpd_exit_fn(FCNAME);
}

#undef FCNAME
#define FCNAME "smpd_priority_class_to_win_class"
int smpd_priority_class_to_win_class(int *priorityClass)
{
    smpd_enter_fn(FCNAME);
    switch (*priorityClass)
    {
    case 0:
	*priorityClass = IDLE_PRIORITY_CLASS;
	break;
    case 1:
	*priorityClass = BELOW_NORMAL_PRIORITY_CLASS;
	break;
    case 2:
	*priorityClass = NORMAL_PRIORITY_CLASS;
	break;
    case 3:
	*priorityClass = ABOVE_NORMAL_PRIORITY_CLASS;
	break;
    case 4:
	*priorityClass = HIGH_PRIORITY_CLASS;
	break;
    default:
	*priorityClass = NORMAL_PRIORITY_CLASS;
	break;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_priority_to_win_priority"
int smpd_priority_to_win_priority(int *priority)
{
    smpd_enter_fn(FCNAME);
    switch (*priority)
    {
    case 0:
	*priority = THREAD_PRIORITY_IDLE;
	break;
    case 1:
	*priority = THREAD_PRIORITY_LOWEST;
	break;
    case 2:
	*priority = THREAD_PRIORITY_BELOW_NORMAL;
	break;
    case 3:
	*priority = THREAD_PRIORITY_NORMAL;
	break;
    case 4:
	*priority = THREAD_PRIORITY_ABOVE_NORMAL;
	break;
    case 5:
	*priority = THREAD_PRIORITY_HIGHEST;
	break;
    default:
	*priority = THREAD_PRIORITY_NORMAL;
	break;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

/* Windows code */

typedef struct smpd_piothread_arg_t
{
    HANDLE hIn;
    SOCKET hOut;
    int pid;
} smpd_piothread_arg_t;

typedef struct smpd_pinthread_arg_t
{
    SOCKET hIn;
    HANDLE hOut;
    int pid;
} smpd_pinthread_arg_t;

#undef FCNAME
#define FCNAME "smpd_easy_send"
static int smpd_easy_send(SOCKET sock, char *buffer, int length)
{
    int error;
    int num_sent, num_left;

    smpd_exit_fn(FCNAME);
    num_left = length;
    while (num_left)
    {
	while ((num_sent = send(sock, buffer, num_left, 0)) == SOCKET_ERROR)
	{
	    error = WSAGetLastError();
	    if (error == WSAEWOULDBLOCK)
	    {
		Sleep(0);
		continue;
	    }
	    if (error == WSAENOBUFS)
	    {
		/* If there is no buffer space available then split the buffer in half and send each piece separately.*/
		if (smpd_easy_send(sock, buffer, num_left/2) == SOCKET_ERROR)
		{
		    smpd_exit_fn(FCNAME);
		    return SOCKET_ERROR;
		}
		if (smpd_easy_send(sock, buffer+(num_left/2), num_left - (num_left/2)) == SOCKET_ERROR)
		{
		    smpd_exit_fn(FCNAME);
		    return SOCKET_ERROR;
		}
		smpd_exit_fn(FCNAME);
		return length;
	    }
	    WSASetLastError(error);
	    smpd_exit_fn(FCNAME);
	    return SOCKET_ERROR;
	}
	num_left = num_left - num_sent;
	buffer = buffer + num_sent;
    }
    smpd_exit_fn(FCNAME);
    return length;
}

int smpd_piothread(smpd_piothread_arg_t *p)
{
    char buffer[8192];
    DWORD num_read;
    HANDLE hIn;
    SOCKET hOut;
    DWORD error;
    char bogus_char;
    int pid;
    double t1, t2;

    hIn = p->hIn;
    hOut = p->hOut;
    pid = p->pid;
    free(p);
    p = NULL;

    smpd_dbg_printf("*** entering smpd_piothread pid:%d sock:%d ***\n", pid, hOut);
    for (;;)
    {
	num_read = 0;
	if (!ReadFile(hIn, buffer, 8192, &num_read, NULL))
	{
	    error = GetLastError();
	    /* If there was an error but some bytes were read, send those bytes before exiting */
	    if (num_read > 0)
	    {
		if (smpd_easy_send(hOut, buffer, num_read) == SOCKET_ERROR)
		{
		    smpd_dbg_printf("smpd_easy_send of %d bytes failed.\n", num_read);
		    break;
		}
	    }
	    smpd_dbg_printf("ReadFile failed, error %d\n", error);
	    break;
	}
	if (num_read < 1)
	{
	    smpd_dbg_printf("ReadFile returned %d bytes\n", num_read);
	    break;
	}
	/*smpd_dbg_printf("*** smpd_piothread read %d bytes ***\n", num_read);*/
	if (smpd_easy_send(hOut, buffer, num_read) == SOCKET_ERROR)
	{
	    smpd_dbg_printf("smpd_easy_send of %d bytes failed.\n", num_read);
	    break;
	}
	/*smpd_dbg_printf("*** smpd_piothread wrote %d bytes ***\n", num_read);*/
    }
    smpd_dbg_printf("*** smpd_piothread finishing pid:%d ***\n", pid);
    /*
    FlushFileBuffers((HANDLE)hOut);
    if (shutdown(hOut, SD_BOTH) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hOut, WSAGetLastError());
    }
    if (closesocket(hOut) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hOut, WSAGetLastError());
    }
    */
    if (shutdown(hOut, SD_SEND) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hOut, WSAGetLastError());
    }
    t1 = PMPI_Wtime();
    recv(hOut, &bogus_char, 1, 0);
    if (closesocket(hOut) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hOut, WSAGetLastError());
    }
    t2 = PMPI_Wtime();
    smpd_dbg_printf("closing output socket took %.3f seconds\n", t2-t1);
    CloseHandle(hIn);
    /*smpd_dbg_printf("*** exiting smpd_piothread ***\n");*/
    return 0;
}

/* one line at a time version */
int smpd_pinthread(smpd_pinthread_arg_t *p)
{
    char str [SMPD_MAX_CMD_LENGTH];
    int index;
    DWORD num_written;
    int num_read;
    SOCKET hIn;
    HANDLE hOut;
    int pid;
    /*int i;*/
    /*
    char bogus_char;
    double t1, t2;
    */

    hIn = p->hIn;
    hOut = p->hOut;
    pid = p->pid;
    free(p);
    p = NULL;

    smpd_dbg_printf("*** entering smpd_pinthread pid:%d sock:%d ***\n", pid, hIn);
    index = 0;
    for (;;)
    {
	num_read = recv(hIn, &str[index], 1, 0);
	if (num_read == SOCKET_ERROR || num_read == 0)
	{
	    if (num_read == SOCKET_ERROR && WSAGetLastError() == WSAEWOULDBLOCK)
	    {
		u_long optval = (u_long)TRUE;
		ioctlsocket(hIn, FIONBIO, &optval);
		continue;
	    }
	    if (index > 0)
	    {
		/* write any buffered data before exiting */
		if (!WriteFile(hOut, str, index, &num_written, NULL))
		{
		    smpd_dbg_printf("WriteFile failed, error %d\n", GetLastError());
		    break;
		}
	    }
	    if (num_read != 0)
	    {
		smpd_dbg_printf("recv from stdin socket failed, error %d.\n", WSAGetLastError());
	    }
	    break;
	}
	if (str[index] == '\n' || index == SMPD_MAX_CMD_LENGTH-1)
	{
	    smpd_dbg_printf("writing %d bytes to the process's stdin\n", index+1);
	    if (!WriteFile(hOut, str, index+1, &num_written, NULL))
	    {
		smpd_dbg_printf("WriteFile failed, error %d\n", GetLastError());
		break;
	    }
	    /*
	    smpd_dbg_printf("wrote: ");
	    for (i=0; i<=index; i++)
	    {
		smpd_dbg_printf("(%d)'%c'", (int)str[i], str[i]);
	    }
	    smpd_dbg_printf("\n");
	    */
	    index = 0;
	}
	else
	{
	    smpd_dbg_printf("read character(%d)'%c'\n", (int)str[index], str[index]);
	    index++;
	}
    }
    smpd_dbg_printf("*** smpd_pinthread finishing pid:%d ***\n", pid);
    FlushFileBuffers(hOut);
    if (shutdown(hIn, SD_BOTH) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hIn, WSAGetLastError());
    }
    if (closesocket(hIn) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hIn, WSAGetLastError());
    }
    /*
    if (shutdown(hIn, SD_SEND) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hIn, WSAGetLastError());
    }
    t1 = PMPI_Wtime();
    recv(hIn, &bogus_char, 1, 0);
    if (closesocket(hIn) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hIn, WSAGetLastError());
    }
    t2 = PMPI_Wtime();
    smpd_dbg_printf("closing input socket took %.3f seconds\n", t2-t1);
    */
    CloseHandle(hOut);
    /*smpd_dbg_printf("*** exiting smpd_pinthread ***\n");*/
    return 0;
}
#if 0
/* 1 byte at a time version */
int smpd_pinthread(smpd_pinthread_arg_t *p)
{
    char ch;
    DWORD num_written;
    SOCKET hIn;
    HANDLE hOut;
    int num_read;

    hIn = p->hIn;
    hOut = p->hOut;
    free(p);
    p = NULL;

    smpd_dbg_printf("*** entering smpd_pinthread ***\n");
    for (;;)
    {
	num_read = recv(hIn, &ch, 1, 0);
	if (num_read == SOCKET_ERROR)
	{
	    if (num_read == SOCKET_ERROR && WSAGetLastError() == WSAEWOULDBLOCK)
	    {
		int optval = TRUE;
		ioctlsocket(hIn, FIONBIO, &optval);
		continue;
	    }
	    smpd_dbg_printf("recv from stdin socket failed, error %d.\n", WSAGetLastError());
	    break;
	}
	if (num_read == 0)
	{
	    break;
	}
	if (!WriteFile(hOut, &ch, 1, &num_written, NULL))
	{
	    smpd_dbg_printf("WriteFile failed, error %d\n", GetLastError());
	    break;
	}
    }
    smpd_dbg_printf("*** smpd_pinthread finishing ***\n");
    FlushFileBuffers(hOut);
    if (closesocket(hIn) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hIn, WSAGetLastError());
    }
    CloseHandle(hOut);
    /*smpd_dbg_printf("*** exiting smpd_pinthread ***\n");*/
    return 0;
}
#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)
{
    HANDLE hStdin = INVALID_HANDLE_VALUE, hStdout = INVALID_HANDLE_VALUE, hStderr = INVALID_HANDLE_VALUE;
    SOCKET hSockStdinR = INVALID_SOCKET, hSockStdinW = INVALID_SOCKET;
    SOCKET hSockStdoutR = INVALID_SOCKET, hSockStdoutW = INVALID_SOCKET;
    SOCKET hSockStderrR = INVALID_SOCKET, hSockStderrW = INVALID_SOCKET;
    SOCKET hSockPmiR = INVALID_SOCKET, hSockPmiW = INVALID_SOCKET;
    HANDLE hPipeStdinR = NULL, hPipeStdinW = NULL;
    HANDLE hPipeStdoutR = NULL, hPipeStdoutW = NULL;
    HANDLE hPipeStderrR = NULL, hPipeStderrW = NULL;
    HANDLE hIn = INVALID_HANDLE_VALUE, hOut = INVALID_HANDLE_VALUE, hErr = INVALID_HANDLE_VALUE;
    STARTUPINFO saInfo;
    PROCESS_INFORMATION psInfo = { 0 };
    void *pEnv=NULL;
    char tSavedPath[MAX_PATH] = ".";
    DWORD launch_flag = 0;
    int nError = 0, result = 0;
    /*unsigned long blocking_flag = 0;*/
    MPIDU_Sock_t sock_in = MPIDU_SOCK_INVALID_SOCK, sock_out = MPIDU_SOCK_INVALID_SOCK, sock_err = MPIDU_SOCK_INVALID_SOCK, sock_pmi = MPIDU_SOCK_INVALID_SOCK;
    SECURITY_ATTRIBUTES saAttr;
    char str[8192], sock_str[20];
    BOOL bSuccess = TRUE;
    char *actual_exe, exe_data[SMPD_MAX_EXE_LENGTH];
    char *args;
    char temp_exe[SMPD_MAX_EXE_LENGTH];
    MPIDU_Sock_t sock_pmi_listener;
    smpd_context_t *listener_context;
    int listener_port = 0;
    char host_description[256];
    char err_msg[MAX_ERROR_LENGTH] = "";

    smpd_enter_fn(FCNAME);

    /* Initialize the psInfo structure in case there is an error an we jump to CLEANUP before psInfo is set */
    psInfo.hProcess = INVALID_HANDLE_VALUE;

    /* resolve the executable name */
    if (process->path[0] != '\0')
    {
	args = process->exe;
	result = MPIU_Str_get_string(&args, 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))
	{
	    if (args != NULL)
	    {
		strncat(exe_data, " ", SMPD_MAX_EXE_LENGTH - strlen(exe_data));
		strncat(exe_data, args, 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;
    }

    smpd_priority_class_to_win_class(&priorityClass);
    smpd_priority_to_win_priority(&priority);

    /* Save stdin, stdout, and stderr */
    result = WaitForSingleObject(smpd_process.hLaunchProcessMutex, INFINITE);
    if (result == WAIT_FAILED)
    {
	result = GetLastError();
	smpd_translate_win_error(result, err_msg, MAX_ERROR_LENGTH, NULL);
	smpd_err_printf("Error waiting for smpd_process.hLaunchProcessMutex %d, %s\n", result, err_msg);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }