smpd_state_machine.c

fortran并行计算包

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#include "smpd.h"

#ifdef HAVE_WINDOWS_H
/*
static void print_bytes(char *str, int length)
{
    int i;
    printf("%d bytes: '", length);
    for (i=0; i<length; i++)
    {
	if (str[i] > 20)
	{
	    printf("%c", str[i]);
	}
	else
	{
	    printf("(%d)", (int)str[i]);
	}
    }
    printf("'\n");
    fflush(stdout);
}
*/
void smpd_stdin_thread(SOCKET hWrite)
{
    DWORD num_read;
    char str[SMPD_MAX_CMD_LENGTH];
    int index;
    HANDLE h = INVALID_HANDLE_VALUE;
    DWORD result;
    char err_msg[256] = "";
    /*char bogus_char;*/

    smpd_dbg_printf("smpd_stdin_thread started.\n");
    /* acquire the launch process mutex to avoid grabbing a redirected input handle */
    result = WaitForSingleObject(smpd_process.hLaunchProcessMutex, INFINITE);
    if (result == WAIT_FAILED)
    {
	result = GetLastError();
	smpd_translate_win_error(result, err_msg, 256, NULL);
	smpd_err_printf("smpd_stdin_thread:WaitForSingleObject(hLaunchProcessMutex) failed: Error %d, %s\n", result, err_msg);
	goto fn_fail;
    }
    h = GetStdHandle(STD_INPUT_HANDLE);
    if (!ReleaseMutex(smpd_process.hLaunchProcessMutex))
    {
	result = GetLastError();
	smpd_translate_win_error(result, err_msg, 256, NULL);
	smpd_err_printf("smpd_stdin_thread:ReleaseMutex(hLaunchProcessMutex) failed: Error %d, %s\n", result, err_msg);
	goto fn_fail;
    }
    if (h == NULL || h == INVALID_HANDLE_VALUE)
    {
	/* Don't print an error in case there is no stdin handle */
	smpd_dbg_printf("Unable to get the stdin handle.\n");
	goto fn_fail;
    }
    index = 0;
    for (;;)
    {
	/*smpd_dbg_printf("waiting for input from stdin\n");*/
	num_read = 0;
	str[index] = '\0';
	if (ReadFile(h, &str[index], 1, &num_read, NULL))
	{
	    if (num_read < 1)
	    {
		/* forward any buffered data */
		if (index > 0)
		{
		    if (send(hWrite, str, index, 0) == SOCKET_ERROR)
		    {
			result = WSAGetLastError();
			smpd_translate_win_error(result, err_msg, 256, NULL);
			smpd_err_printf("unable to forward stdin, send failed, error %d, %s\n", result, err_msg);
			goto fn_fail;
		    }
		}
		/* ReadFile failed, what do I do? */
		smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
		goto fn_fail;
	    }
	    /*printf("CHAR(%d)", (int)str[index]);fflush(stdout);*/
	    if (str[index] == '\n' || index == SMPD_MAX_CMD_LENGTH-1)
	    {
		num_read = index + 1;
		index = 0; /* reset the index back to the beginning of the input buffer */
		smpd_dbg_printf("forwarding stdin: %d bytes\n", num_read);
		/*print_bytes(str, num_read);*/
		if (send(hWrite, str, num_read, 0) == SOCKET_ERROR)
		{
		    result = WSAGetLastError();
		    smpd_translate_win_error(result, err_msg, 256, NULL);
		    smpd_err_printf("unable to forward stdin, send failed, error %d, %s\n", result, err_msg);
		    goto fn_fail;
		}
	    }
	    else
	    {
		index++;
	    }
	}
	else
	{
	    /* forward any buffered data */
	    if (index > 0)
	    {
		if (send(hWrite, str, index, 0) == SOCKET_ERROR)
		{
		    result = WSAGetLastError();
		    smpd_translate_win_error(result, err_msg, 256, NULL);
		    smpd_err_printf("unable to forward stdin, send failed, error %d, %s\n", result, err_msg);
		    goto fn_fail;
		}
	    }
	    /* ReadFile failed, what do I do? */
	    smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
	    goto fn_fail;
	}
    }
fn_exit:
    return;
fn_fail:
    /* graceful shutdown
    if (shutdown(hWrite, SD_SEND) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
    }
    recv(hWrite, &bogus_char, 1, 0);
    if (closesocket(hWrite) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
    }
    */
    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
    {
	smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
    }
    if (closesocket(hWrite) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
    }
    goto fn_exit;
}
/* forward entire input lines at a time version */
#if 0
void smpd_stdin_thread(SOCKET hWrite)
{
    DWORD num_read;
    char str[SMPD_MAX_CMD_LENGTH];
    int index;
    HANDLE h[2];
    int result;
    /*int i;*/

    smpd_dbg_printf("smpd_stdin_thread started.\n");
    /* acquire the launch process mutex to avoid grabbing a redirected input handle */
    WaitForSingleObject(smpd_process.hLaunchProcessMutex, INFINITE);
    h[0] = GetStdHandle(STD_INPUT_HANDLE);
    ReleaseMutex(smpd_process.hLaunchProcessMutex);
    if (h[0] == NULL || h[0] == INVALID_HANDLE_VALUE)
    {
	smpd_err_printf("Unable to get the stdin handle.\n");
	return;
    }
    h[1] = smpd_process.hCloseStdinThreadEvent;
    index = 0;
    for (;;)
    {
	/*smpd_dbg_printf("waiting for input from stdin\n");*/
	result = WaitForMultipleObjects(2, h, FALSE, INFINITE);
	if (result == WAIT_OBJECT_0)
	{
	    num_read = 0;
	    str[index] = '\0';
	    if (ReadFile(h[0], &str[index], 1, &num_read, NULL))
	    {
		if (num_read < 1)
		{
		    /* ReadFile failed, what do I do? */
		    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
		    {
			smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		    }
		    if (closesocket(hWrite) == SOCKET_ERROR)
		    {
			smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		    }
		    smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
		    return;
		}
		/*printf("CHAR(%d)", (int)str[index]);fflush(stdout);*/
		if (str[index] == '\n' || index == SMPD_MAX_CMD_LENGTH-1)
		{
		    num_read = index + 1;
		    index = 0; /* reset the index back to the beginning of the input buffer */
		    smpd_dbg_printf("forwarding stdin: %d bytes\n", num_read);
		    print_bytes(str, num_read);
		    /*
		    printf("forwarding stdin: '");
		    for (i=0; i<num_read; i++)
		    {
		    printf("%c", str[i]);
		    }
		    printf("'\n");
		    fflush(stdout);
		    */
		    if (send(hWrite, str, num_read, 0) == SOCKET_ERROR)
		    {
			smpd_err_printf("unable to forward stdin, send failed, error %d\n", WSAGetLastError());
			return;
		    }
		}
		else
		{
		    index++;
		}
	    }
	    else
	    {
		/* ReadFile failed, what do I do? */
		if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
		{
		    smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		if (closesocket(hWrite) == SOCKET_ERROR)
		{
		    smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
		return;
	    }
	}
	else if (result == WAIT_OBJECT_0 + 1)
	{
	    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
	    {
		smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    if (closesocket(hWrite) == SOCKET_ERROR)
	    {
		smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    smpd_dbg_printf("hCloseStdinThreadEvent signalled, closing stdin reader thread.\n");
	    return;
	}
	else
	{
	    smpd_err_printf("stdin wait failed, error %d\n", GetLastError());
	    return;
	}
    }
}
#endif
#if 0
/* 1 byte at a time version */
void smpd_stdin_thread(SOCKET hWrite)
{
    DWORD num_read;
    char str[SMPD_MAX_CMD_LENGTH];
    HANDLE h[2];
    int result;
    /*int i;*/

    smpd_dbg_printf("smpd_stdin_thread started.\n");
    /* acquire the launch process mutex to avoid grabbing a redirected input handle */
    WaitForSingleObject(smpd_process.hLaunchProcessMutex, INFINITE);
    h[0] = GetStdHandle(STD_INPUT_HANDLE);
    ReleaseMutex(smpd_process.hLaunchProcessMutex);
    if (h[0] == NULL)
    {
	smpd_err_printf("Unable to get the stdin handle.\n");
	return;
    }
    h[1] = smpd_process.hCloseStdinThreadEvent;
    for (;;)
    {
	/*smpd_dbg_printf("waiting for input from stdin\n");*/
	result = WaitForMultipleObjects(2, h, FALSE, INFINITE);
	if (result == WAIT_OBJECT_0)
	{
	    if (ReadFile(h[0], str, 1/*SMPD_MAX_CMD_LENGTH*/, &num_read, NULL))
	    {
		smpd_dbg_printf("forwarding stdin: %d bytes\n", num_read);
		/*
		printf("forwarding stdin: '");
		for (i=0; i<num_read; i++)
		{
		    printf("%c", str[i]);
		}
		printf("'\n");
		fflush(stdout);
		*/
		if (num_read > 0)
		{
		    if (send(hWrite, str, num_read, 0) == SOCKET_ERROR)
		    {
			smpd_err_printf("unable to forward stdin, send failed, error %d\n", WSAGetLastError());
			return;
		    }
		}
		else
		{
		    /* ReadFile failed, what do I do? */
		    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
		    {
			smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		    }
		    if (closesocket(hWrite) == SOCKET_ERROR)
		    {
			smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		    }
		    smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
		    return;
		}
	    }
	    else
	    {
		/* ReadFile failed, what do I do? */
		if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
		{
		    smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		if (closesocket(hWrite) == SOCKET_ERROR)
		{
		    smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		smpd_dbg_printf("ReadFile failed, closing stdin reader thread.\n");
		return;
	    }
	}
	else if (result == WAIT_OBJECT_0 + 1)
	{
	    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
	    {
		smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    if (closesocket(hWrite) == SOCKET_ERROR)
	    {
		smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    smpd_dbg_printf("hCloseStdinThreadEvent signalled, closing stdin reader thread.\n");
	    return;
	}
	else
	{
	    smpd_err_printf("stdin wait failed, error %d\n", GetLastError());
	    return;
	}
    }
}
#endif
#if 0
/* fgets version */
void smpd_stdin_thread(SOCKET hWrite)
{
    DWORD len;
    char str[SMPD_MAX_CMD_LENGTH];
    HANDLE h[2];
    int result;

    smpd_dbg_printf("smpd_stdin_thread started.\n");
    /* acquire the launch process mutex to avoid grabbing a redirected input handle */
    WaitForSingleObject(smpd_process.hLaunchProcessMutex, INFINITE);
    h[0] = GetStdHandle(STD_INPUT_HANDLE);
    ReleaseMutex(smpd_process.hLaunchProcessMutex);
    if (h[0] == NULL)
    {
	smpd_err_printf("Unable to get the stdin handle.\n");
	return;
    }