redirectio.c

刚才是说明 现在是安装程序在 LINUX环境下进行编程的MPICH安装文件

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*  $Id: redirectio.c,v 1.1 2002/10/07 22:12:45 toonen Exp $
 *
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#include "pmiimpl.h"
#include <stdio.h>
#include "bsocket.h"
#include "mpdutil.h"
#include "redirectio.h"

#ifndef dbg_printf
#define dbg_printf printf
#endif
#ifndef err_printf
#define err_printf printf
#endif

static int g_bfdListen;
static HANDLE g_hListenReleasedEvent;

void InitRedirection()
{
    g_hListenReleasedEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
}

void FinalizeRedirection()
{
    CloseHandle(g_hListenReleasedEvent);
}

static void RedirectStdin(int bfd)
{
    char pBuffer[1024];
    DWORD num_read;
    HANDLE hStdin;
    int i;

    for (i=0; i<3; i++)
    {
	hStdin = GetStdHandle(STD_INPUT_HANDLE);
	if (hStdin != INVALID_HANDLE_VALUE)
	    break;
	Sleep(10);
    }
    if (hStdin == INVALID_HANDLE_VALUE)
    {
	err_printf("Critical error: Unable to acquire the standard input handle for redirection. error %d\n", GetLastError());
	fflush(stdout);
    }

    while (ReadFile(hStdin, pBuffer, 1024, &num_read, NULL))
    {
	if (beasy_send(bfd, pBuffer, num_read) == SOCKET_ERROR)
	    break;
    }
    beasy_closesocket(bfd);
}

void RedirectIOThread2(int abort_bfd)
{
    int client_bfd, child_abort_bfd = BFD_INVALID_SOCKET;
    int bfdListen;
    int n, i;
    char pBuffer[1024];
    DWORD num_read, num_written;
    bfd_set total_set, readset;
    int bfdActive[FD_SETSIZE];
    int nActive = 0;
    HANDLE hStdout, hStderr, hOut;
    int nRank;
    char cType;
    int nDatalen;
    BOOL bDeleteOnEmpty = FALSE;
    HANDLE hChildThread = NULL;
    DWORD dwThreadId;
    HANDLE hThread;

    hStdout = GetStdHandle(STD_OUTPUT_HANDLE);
    hStderr = GetStdHandle(STD_ERROR_HANDLE);
    
    BFD_ZERO(&total_set);
    BFD_SET(abort_bfd, &total_set);
    BFD_SET(g_bfdListen, &total_set);
    bfdListen = g_bfdListen;

    while (TRUE)
    {
	readset = total_set;
	n = bselect(0, &readset, NULL, NULL, NULL);
	if (n == SOCKET_ERROR)
	{
	    err_printf("RedirectIOThread2: bselect failed, error %d\n", WSAGetLastError());fflush(stdout);
	    beasy_closesocket(abort_bfd);
	    for (i=0; i<nActive; i++)
		beasy_closesocket(bfdActive[i]);
	    if (bfdListen != BFD_INVALID_SOCKET)
	    {
		SetEvent(g_hListenReleasedEvent);
		/*beasy_closesocket(bfdListen);*/
	    }
	    if (hChildThread != NULL)
		CloseHandle(hChildThread);
	    return;
	}
	if (n == 0)
	{
	    err_printf("RedirectIOThread2: bselect returned zero sockets available\n");fflush(stdout);
	    beasy_closesocket(abort_bfd);
	    for (i=0; i<nActive; i++)
		beasy_closesocket(bfdActive[i]);
	    if (bfdListen != BFD_INVALID_SOCKET)
	    {
		SetEvent(g_hListenReleasedEvent);
		/*beasy_closesocket(bfdListen);*/
	    }
	    if (hChildThread != NULL)
		CloseHandle(hChildThread);
	    return;
	}
	else
	{
	    if (BFD_ISSET(abort_bfd, &readset))
	    {
		char c;
		BOOL bCloseNow = TRUE;
		num_read = beasy_receive(abort_bfd, &c, 1);
		if (num_read == 1)
		{
		    if (c == 0)
		    {
			if (child_abort_bfd != BFD_INVALID_SOCKET)
			    beasy_send(child_abort_bfd, &c, 1);

			if (nActive == 0)
			    WaitForSingleObject(hChildThread, 10000);
			else
			    bCloseNow = FALSE;
			bDeleteOnEmpty = TRUE;
		    }
		}
		if (bCloseNow)
		{
		    for (i=0; i<nActive; i++)
		    {
			beasy_closesocket(bfdActive[i]);
		    }
		    nActive = 0;
		    if (child_abort_bfd == BFD_INVALID_SOCKET)
			SetEvent(g_hListenReleasedEvent);
		    else
		    {
			beasy_send(child_abort_bfd, "x", 1);
			beasy_closesocket(child_abort_bfd);
		    }
		    beasy_closesocket(abort_bfd);
		    if (hChildThread != NULL)
			CloseHandle(hChildThread);
		    return;
		}
	    }
	    if (bfdListen != BFD_INVALID_SOCKET && BFD_ISSET(bfdListen, &readset))
	    {
		if ((nActive + 3) >= FD_SETSIZE)
		{
		    int temp_bfd;
		    MakeLoop(&temp_bfd, &child_abort_bfd);
		    if (temp_bfd == BFD_INVALID_SOCKET || child_abort_bfd == BFD_INVALID_SOCKET)
		    {
			err_printf("Critical error: Unable to create a socket.\n");fflush(stdout);
			break;
		    }
		    hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RedirectIOThread2, (LPVOID)temp_bfd, 0, &dwThreadId);
		    if (hThread == NULL)
		    {
			err_printf("Critical error: Unable to create an io thread\n");fflush(stdout);
			break;
		    }
		    CloseHandle(hThread);
		    BFD_CLR(bfdListen, &total_set);
		    bfdListen = BFD_INVALID_SOCKET;
		}
		else
		{
		    client_bfd = beasy_accept(bfdListen);
		    if (client_bfd == BFD_INVALID_SOCKET)
		    {
			int error = WSAGetLastError();
			err_printf("RedirectIOThread2: baccept failed: %d\n", error);fflush(stdout);
			break;
		    }
		    
		    if (beasy_receive(client_bfd, &cType, sizeof(char)) == SOCKET_ERROR)
		    {
			break;
		    }
		    
		    if (cType == 0)
		    {
			DWORD dwThreadID;
			hChildThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RedirectStdin, (void*)client_bfd, 0, &dwThreadID);
		    }
		    else
		    {
			bfdActive[nActive] = client_bfd;
			BFD_SET(client_bfd, &total_set);
			nActive++;
		    }
		}
		n--;
	    }
	    if (n > 0)
	    {
		for (i=0; n > 0; i++)
		{
		    if (BFD_ISSET(bfdActive[i], &readset))
		    {
			char pTemp[sizeof(int)+sizeof(char)+sizeof(int)];
			num_read = beasy_receive(bfdActive[i], pTemp, sizeof(int)+sizeof(char)+sizeof(int));
			if (num_read == SOCKET_ERROR || num_read == 0)
			{
			    BFD_CLR(bfdActive[i], &total_set);
			    beasy_closesocket(bfdActive[i]);
			    nActive--;
			    bfdActive[i] = bfdActive[nActive];
			    i--;
			    /*dbg_printf("(-%d)", nActive);fflush(stdout);*/
			}
			else
			{
			    nDatalen = *(int*)pTemp;
			    cType = pTemp[sizeof(int)];
			    nRank = *(int*)&pTemp[sizeof(int)+sizeof(char)];
			    num_read = beasy_receive(bfdActive[i], pBuffer, nDatalen);
			    if (num_read == SOCKET_ERROR || num_read == 0)
			    {
				BFD_CLR(bfdActive[i], &total_set);
				beasy_closesocket(bfdActive[i]);
				nActive--;
				bfdActive[i] = bfdActive[nActive];
				i--;
				/*dbg_printf("(-%d)", nActive);fflush(stdout);*/
			    }
			    else
			    {
				hOut = (cType == 1) ? hStdout : hStderr;
				/*
				if (g_bDoMultiColorOutput)
				{
				    WaitForSingleObject(g_hConsoleOutputMutex, INFINITE);
				    SetConsoleTextAttribute(hOut, aConsoleColorAttribute[nRank%NUM_OUTPUT_COLORS]);
				    if (WriteFile(hOut, pBuffer, num_read, &num_written, NULL))
					FlushFileBuffers(hOut);
				    else
				    {
					err_printf("*** output lost ***\n");fflush(stdout);
				    }
				    SetConsoleTextAttribute(hOut, g_ConsoleAttribute);
				    ReleaseMutex(g_hConsoleOutputMutex);
				}
				else
				{
				*/
				    if (!WriteFile(hOut, pBuffer, num_read, &num_written, NULL))
				    {
					err_printf("*** output lost ***\n");fflush(stdout);
				    }
				/*}*/
			    }
			}
			n--;
		    }
		}
	    }
	    if (bDeleteOnEmpty && nActive == 0)
	    {
		if (hChildThread != NULL)
		{
		    WaitForSingleObject(hChildThread, 10000);
		    CloseHandle(hChildThread);
		    hChildThread = NULL;
		}
		break;
	    }
	}
    }

    for (i=0; i<nActive; i++)
    {
	beasy_closesocket(bfdActive[i]);
    }