redirectio.cpp

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

#include "stdafx.h"
#include "guiMPIRun.h"
#include "guiMPIRunDoc.h"
#include "guiMPIRunView.h"
#include "RedirectIO.h"
#include "global.h"
#include <stdio.h>
#include "mpdutil.h"

static CGuiMPIRunView *g_pDlg;
static HANDLE g_hConsoleOutputMutex;
static SOCKET g_sockListen;
static HANDLE g_hListenReleasedEvent = CreateEvent(NULL, TRUE, FALSE, NULL);

static void WriteOutputToRichEdit(char *pszStr, COLORREF color, CRichEditCtrl &edit)
{
    CHARFORMAT cf;
    int last;
    int nBefore;

    nBefore = edit.GetLineCount();

    cf.dwMask = CFM_COLOR;
    cf.crTextColor = color;
    cf.dwEffects = 0;
    last = -1;
    edit.SetSel(last, last);
    edit.ReplaceSel(pszStr);
    last = edit.GetTextLength();
    edit.SetSel(last - strlen(pszStr), last);
    edit.SetSelectionCharFormat(cf);
    edit.SetSel(-1, -1);
    edit.LineScroll(edit.GetLineCount() - nBefore);

    if (g_pDlg->m_redirect && g_pDlg->m_fout)
    {
	fprintf(g_pDlg->m_fout, "%s", pszStr);
	fflush(g_pDlg->m_fout);
    }
}

void RedirectRichEdit(SOCKET sock)
{
    HANDLE hEvent[3];
    DWORD dwResult;

    hEvent[0] = g_pDlg->m_hAbortEvent;
    hEvent[1] = g_pDlg->m_hJobFinished;
    hEvent[2] = g_pDlg->m_hRedirectStdinEvent;

    while (true)
    {
	dwResult = WaitForMultipleObjects(3, hEvent, FALSE, INFINITE);
	switch (dwResult)
	{
	case WAIT_OBJECT_0:
	case WAIT_OBJECT_0+1:
	    easy_closesocket(sock);
	    CloseHandle(g_pDlg->m_hRedirectRicheditThread);
	    g_pDlg->m_hRedirectRicheditThread = NULL;
	    return;
	    break;
	case WAIT_OBJECT_0+2:
	    if (WaitForSingleObject(g_pDlg->m_hRedirectStdinMutex, 10000) == WAIT_OBJECT_0)
	    {
		if (g_pDlg->m_pRedirectStdinList == NULL)
		{
		    easy_closesocket(sock);
		    ReleaseMutex(g_pDlg->m_hRedirectStdinMutex);
		    CloseHandle(g_pDlg->m_hRedirectRicheditThread);
		    g_pDlg->m_hRedirectRicheditThread = NULL;
		    return;
		}
		if (easy_send(sock, g_pDlg->m_pRedirectStdinList->str.GetBuffer(0), g_pDlg->m_pRedirectStdinList->str.GetLength()) == SOCKET_ERROR)
		{
		    easy_closesocket(sock);
		    ReleaseMutex(g_pDlg->m_hRedirectStdinMutex);
		    CloseHandle(g_pDlg->m_hRedirectRicheditThread);
		    g_pDlg->m_hRedirectRicheditThread = NULL;
		    return;
		}
		CGuiMPIRunView::RedirectStdinStruct *pNode = g_pDlg->m_pRedirectStdinList;
		g_pDlg->m_pRedirectStdinList = g_pDlg->m_pRedirectStdinList->pNext;
		delete pNode;
		if (g_pDlg->m_pRedirectStdinList == NULL)
		    ResetEvent(g_pDlg->m_hRedirectStdinEvent);
		ReleaseMutex(g_pDlg->m_hRedirectStdinMutex);
	    }
	    break;
	default:
	    easy_closesocket(sock);
	    CloseHandle(g_pDlg->m_hRedirectRicheditThread);
	    g_pDlg->m_hRedirectRicheditThread = NULL;
	    return;
	    break;
	}
    }
}

void RedirectIOThread2(SOCKET abort_sock)
{
    SOCKET client_sock, child_abort_sock = INVALID_SOCKET;
    int n, i;
#ifdef USE_LINGER_SOCKOPT
    struct linger linger;
#endif
    BOOL b;
    char pBuffer[1024];
    DWORD num_read;
    fd_set total_set, readset;
    SOCKET sockActive[FD_SETSIZE];
    int nActive = 0;
    int nRank;
    char cType;
    int nDatalen;
    bool bDeleteOnEmpty = false;
    HANDLE hChildThread = NULL;
    
    FD_ZERO(&total_set);
    FD_SET(abort_sock, &total_set);
    FD_SET(g_sockListen, &total_set);

    while (true)
    {
	readset = total_set;
	n = select(0, &readset, NULL, NULL, NULL);
	if (n == SOCKET_ERROR)
	{
	    printf("RedirectIOControlThread2: bselect failed, error %d\n", WSAGetLastError());fflush(stdout);
	    easy_closesocket(abort_sock);
	    for (i=0; i<nActive; i++)
		easy_closesocket(sockActive[i]);
	    return;
	}
	if (n == 0)
	{
	    printf("RedirectIOControlThread2: bselect returned zero sockets available\n");fflush(stdout);
	    easy_closesocket(abort_sock);
	    for (i=0; i<nActive; i++)
		easy_closesocket(sockActive[i]);
	    return;
	}
	else
	{
	    if (FD_ISSET(abort_sock, &readset))
	    {
		char c;
		bool bCloseNow = true;
		num_read = easy_receive(abort_sock, &c, 1);
		if (num_read == 1)
		{
		    if (c == 0)
		    {
			if (child_abort_sock != INVALID_SOCKET)
			    easy_send(child_abort_sock, &c, 1);

			if (nActive == 0)
			    WaitForSingleObject(hChildThread, 10000);
			else
			    bCloseNow = false;
			bDeleteOnEmpty = true;
		    }
		}
		if (bCloseNow)
		{
		    for (i=0; i<nActive; i++)
			easy_closesocket(sockActive[i]);
		    nActive = 0;
		    if (child_abort_sock == INVALID_SOCKET)
			SetEvent(g_hListenReleasedEvent);
		    else
		    {
			easy_send(child_abort_sock, "x", 1);
			easy_closesocket(child_abort_sock);
		    }
		    easy_closesocket(abort_sock);
		    if (hChildThread != NULL)
			CloseHandle(hChildThread);
		    return;
		}
	    }
	    if (FD_ISSET(g_sockListen, &readset))
	    {
		if ((nActive + 3) >= FD_SETSIZE)
		{
		    SOCKET temp_sock;
		    MakeLoop(&temp_sock, &child_abort_sock);
		    if (temp_sock == INVALID_SOCKET || child_abort_sock == INVALID_SOCKET)
		    {
			MessageBox(NULL, "Unable to create a socket", "Critical error", MB_OK);
			break;
		    }
		    DWORD dwThreadId;
		    hChildThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RedirectIOThread2, (LPVOID)temp_sock, 0, &dwThreadId);
		    if (hChildThread == NULL)
		    {
			MessageBox(NULL, "Unable to create an io thread", "Critical error", MB_OK);
			break;
		    }
		    FD_CLR(g_sockListen, &total_set);
		}
		else
		{
		    client_sock = easy_accept(g_sockListen);
		    if (client_sock == INVALID_SOCKET)
		    {
			char str[256];
			int error = WSAGetLastError();
			sprintf(str, "RedirectIOControlThread: baccept failed: %d\n", error);
			MessageBox(NULL, str, "Error", MB_OK);
			break;
		    }
#ifdef USE_LINGER_SOCKOPT
		    linger.l_onoff = 1;
		    linger.l_linger = 60;
		    if (setsockopt(client_sock, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger)) == SOCKET_ERROR)
		    {
			char str[256];
			int error = WSAGetLastError();
			sprintf(str, "RedirectIOControlThread: bsetsockopt failed: %d\n", error);
			MessageBox(NULL, str, "Error", MB_OK);
			easy_closesocket(client_sock);
			break;
		    }
#endif
		    b = TRUE;
		    setsockopt(client_sock, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));
		    
		    if (easy_receive(client_sock, &cType, sizeof(char)) == SOCKET_ERROR)
			break;
		    
		    if (cType == 0)
		    {
			DWORD dwThreadID;
			if (g_pDlg->m_hRedirectRicheditThread != NULL)
			    TerminateThread(g_pDlg->m_hRedirectRicheditThread, 0);
			g_pDlg->m_hRedirectRicheditThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RedirectRichEdit, (void*)client_sock, 0, &dwThreadID);
		    }
		    else
		    {
			sockActive[nActive] = client_sock;
			FD_SET(client_sock, &total_set);
			nActive++;
		    }
		}
		n--;
	    }
	    if (n > 0)
	    {
		for (i=0; n > 0; i++)
		{
		    if (FD_ISSET(sockActive[i], &readset))
		    {
			char pTemp[sizeof(int)+sizeof(char)+sizeof(int)];
			num_read = easy_receive(sockActive[i], pTemp, sizeof(int)+sizeof(char)+sizeof(int));
			if (num_read == SOCKET_ERROR || num_read == 0)
			{
			    FD_CLR(sockActive[i], &total_set);
			    easy_closesocket(sockActive[i]);
			    nActive--;
			    sockActive[i] = sockActive[nActive];
			    i--;
			    //printf("(-%d)", nActive);fflush(stdout);
			}
			else
			{
			    nDatalen = *(int*)pTemp;
			    cType = pTemp[sizeof(int)];
			    nRank = *(int*)&pTemp[sizeof(int)+sizeof(char)];
			    num_read = easy_receive(sockActive[i], pBuffer, nDatalen);
			    if (num_read == SOCKET_ERROR || num_read == 0)
			    {
				FD_CLR(sockActive[i], &total_set);
				easy_closesocket(sockActive[i]);
				nActive--;
				sockActive[i] = sockActive[nActive];
				i--;
				//printf("(-%d)", nActive);fflush(stdout);
			    }
			    else
			    {