mpdleft.cpp

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

#include "mpdimpl.h"
#include "mpdutil.h"
#include <string.h>
#include <winsock2.h>
#include <windows.h>
#include "Service.h"
#include "GetStringOpt.h"
#include "database.h"
#include "Translate_Error.h"
#include <time.h>

struct TmpFileStruct
{
    char pszFileName[MAX_PATH];
    TmpFileStruct *pNext;
};

TmpFileStruct *g_pTmpFileList = NULL;

void statTmp(char *pszOutput, int length)
{
    TmpFileStruct *p;

    *pszOutput = '\0';
    length--; // leave room for the null character

    if (g_pTmpFileList == NULL)
	return;

    if (!snprintf_update(pszOutput, length, "TMP FILES:\n"))
	return;

    p = g_pTmpFileList;
    while (p)
    {
	if (!snprintf_update(pszOutput, length, " '%s'\n", p->pszFileName))
	    return;
	p = p->pNext;
    }
}

static void CreateTmpFile(char *pszFileName, bool bDelete = true)
{
    char pszDir[MAX_PATH] = "C:\\";
    char pszTemp[MAX_PATH];
    char *namepart;
    
    if (!ReadMPDRegistry("temp", pszDir))
	dbg_printf("no temp directory specified, using c:\\\n");
    
    if (GetTempFileName(pszDir, "mpi", 0, pszTemp) == 0)
    //if (GetTempFileName(g_pszTempDir, "mpi", 0, pszTemp) == 0)
    {
	int nError = GetLastError();
	Translate_Error(nError, pszFileName, "FAIL ");
	err_printf("GetTempFileName(%s) failed, %s", pszDir, pszFileName);
	return;
    }
    
    GetFullPathName(pszTemp, MAX_PATH, pszFileName, &namepart);

    if (bDelete)
    {
	// Add this name to the global list
	// These names will be matched with corresponding "deletetmpfile" commands
	// All remaining files will be deleted when the mpd exits
	TmpFileStruct *pNode = new TmpFileStruct;
	strncpy(pNode->pszFileName, pszFileName, MAX_PATH);
	pNode->pNext = g_pTmpFileList;
	g_pTmpFileList = pNode;
    }
}

static bool DeleteTmpFile(char *pszFileName)
{
    TmpFileStruct *pNode, *pTrailer;
    pTrailer = pNode = g_pTmpFileList;
    while (pNode)
    {
	if (stricmp(pszFileName, pNode->pszFileName) == 0)
	{
	    if (pNode == g_pTmpFileList)
	    {
		g_pTmpFileList = g_pTmpFileList->pNext;
		delete pNode;
	    }
	    else
	    {
		pTrailer->pNext = pNode->pNext;
		delete pNode;
	    }
	    return (DeleteFile(pszFileName) == TRUE);
	}
	if (pTrailer != pNode)
	    pTrailer = pTrailer->pNext;
	pNode = pNode->pNext;
    }
    return false;
}

void RemoveAllTmpFiles()
{
    TmpFileStruct *pNode;
    while (g_pTmpFileList)
    {
	pNode = g_pTmpFileList;
	g_pTmpFileList = g_pTmpFileList->pNext;
	if (strlen(pNode->pszFileName) > 0)
	    DeleteFile(pNode->pszFileName);
	delete pNode;
    }
}

#define DEFAULT_MPICH_ROOT_TIMEOUT 7

static int GetPortFromFile(char *pszFileName, int nPid, int *nPort)
{
    int nError;
    DWORD num_read = 0;
    char pBuffer[100] = "";
    char *pChar = pBuffer;
    clock_t cStart;
    HANDLE hProcess = NULL;
    DWORD dwExitCode;
    int nTimeout = DEFAULT_MPICH_ROOT_TIMEOUT;

    if (ReadMPDRegistry("timeout", pBuffer))
    {
	nTimeout = atoi(pBuffer);
	if (nTimeout > 1000)
	    nTimeout = nTimeout / 1000;
	if (nTimeout < 1)
	    nTimeout = 1;
	pBuffer[0] = '\0';
    }

    HANDLE hFile = CreateFile(pszFileName, GENERIC_READ, FILE_SHARE_WRITE, NULL, OPEN_EXISTING, 0, NULL);
    if (hFile == INVALID_HANDLE_VALUE)
    {
	nError = GetLastError();
	return nError;
    }
    
    cStart = clock();
    while (true)
    {
	num_read = 0;
	if (!ReadFile(hFile, pChar, 100, &num_read, NULL))
	{
	    nError = GetLastError();
	    CloseHandle(hFile);
	    DeleteTmpFile(pszFileName);
	    if (hProcess)
		CloseHandle(hProcess);
	    return nError;
	}
	if (num_read == 0)
	{
	    if (!hProcess)
	    {
		hProcess = OpenProcess(PROCESS_QUERY_INFORMATION, FALSE, nPid);
		if (hProcess == NULL)
		{
		    int error = GetLastError();
		    if (error == ERROR_INVALID_PARAMETER)
		    {
			int nExitCode = ConsoleGetExitCode(nPid);
			if (nExitCode != -1 && nExitCode != -2)
			{
			    if (nExitCode == ERROR_WAIT_NO_CHILDREN)
				return -2;
			    *nPort = nExitCode;
			    return -3;
			}
		    }
		}
	    }
	    if (hProcess)
	    {
		if (GetExitCodeProcess(hProcess, &dwExitCode))
		{
		    if (dwExitCode != STILL_ACTIVE)
		    {
			CloseHandle(hProcess);
			if (dwExitCode == ERROR_WAIT_NO_CHILDREN)
			    return -2;
			*nPort = dwExitCode;
			return -3;
		    }
		}
	    }
	    if (clock() - cStart > nTimeout * CLOCKS_PER_SEC)
	    {
		CloseHandle(hFile);
		DeleteTmpFile(pszFileName);
		if (hProcess)
		    CloseHandle(hProcess);
		return -1;
	    }
	    Sleep(100);
	}
	else
	{
	    for (unsigned int i=0; i<num_read; i++)
	    {
		if (*pChar == '\n')
		    break;
		pChar ++;
	    }
	    if (*pChar == '\n')
		break;
	}
    }
    CloseHandle(hFile);
    DeleteTmpFile(pszFileName);
    
    *nPort = atoi(pBuffer);
    
    if (hProcess)
	CloseHandle(hProcess);
    return 0;
}

static void HandleDBCommandRead(MPD_Context *p)
{
    char name[MAX_DBS_NAME_LEN+1];
    char key[MAX_DBS_KEY_LEN+1];
    char value[MAX_DBS_VALUE_LEN+1];
    char pszStr[MAX_CMD_LENGTH];

    char pszSrc[100] = "";
    char pszBfd[10] = "";
    char *pszCmdData;

    SOCKET sock;

    GetStringOpt(p->pszIn, "src", pszSrc);
    //GetStringOpt(p->pszIn, "sock", pszBfd); // don't use GetStringOpt because the data after sock= may be too long
    pszCmdData = strstr(p->pszIn, "sock=");
    pszCmdData += 5; // length of string "sock="
    sock = atoi(pszCmdData);
    sprintf(pszBfd, "%d", sock);
    while (*pszCmdData != ' ')
	pszCmdData++;
    pszCmdData++;

    //dbg_printf("left - HandleDBCommand: src='%s' sock='%s' data='%s'\n", pszSrc, pszBfd, pszCmdData);

    if ((stricmp(pszSrc, g_pszHost) == 0) || (strcmp(pszSrc, g_pszIP) == 0))
    {
	// Stop result-less commands
	if ((strnicmp(p->pszIn, "dbcreate ", 9) == 0) ||
	    (strnicmp(p->pszIn, "dbdestroy ", 10) == 0) ||
	    (strnicmp(p->pszIn, "dbfirst ", 8) == 0))
	{
	    return;
	}

	// Handle the full ring commands
	if (strnicmp(p->pszIn, "dbnext ", 7) == 0)
	{
	    ContextWriteString(GetContext(atoi(pszBfd)), DBS_END_STR);
	    return;
	}

	// The command has gone full circle without succeeding
	if (p = GetContext(sock))
	    ContextWriteString(p, DBS_FAIL_STR);
	else
	{
	    err_printf("GetContext failed for '%s'\n", pszBfd);
	}
	return;
    }

    if (strnicmp(p->pszIn, "dbresult ", 9) == 0)
    {
	char pszDest[MAX_HOST_LENGTH] = "";
	GetStringOpt(p->pszIn, "dest", pszDest);
	if ((stricmp(pszDest, g_pszHost) == 0) || (strcmp(pszDest, g_pszIP) == 0))
	{
	    char *token;
	    token = strstr(p->pszIn, "result=");
	    if (token != NULL)
	    {
		token = &token[7];
		p = GetContext(sock);
		if (p)
		    ContextWriteString(p, token);
		else
		{
		    err_printf("GetContext failed for '%s'\n", pszBfd);
		}
	    }
	    else
	    {
		err_printf("'result=' not found in dbresult command\n");
	    }
	}
	else
	{
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "dbget ", 6) == 0)
    {
	GetNameKeyValue(pszCmdData, name, key, NULL);
	if (dbs_get(name, key, value) == DBS_SUCCESS)
	{
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbresult dest=%s sock=%s result=%s", pszSrc, pszBfd, value);
	    ContextWriteString(g_pRightContext, pszStr);
	}
	else
	{
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "dbcreate ", 9) == 0)
    {
	if (GetStringOpt(p->pszIn, "name", name))
	{
	    dbs_create_name_in(name);
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
	else
	{
	    err_printf("This cannot happen because it should have been caught at the source host\n");
	}
    }
    else if (strnicmp(p->pszIn, "dbdestroy ", 10) == 0)
    {
	GetNameKeyValue(pszCmdData, name, NULL, NULL);
	dbs_destroy(name);
	ContextWriteString(g_pRightContext, p->pszIn);
    }
    else if (strnicmp(p->pszIn, "dbfirst ", 8) == 0)
    {
	GetNameKeyValue(pszCmdData, name, NULL, NULL);
	dbs_first(name, NULL, NULL);
	ContextWriteString(g_pRightContext, p->pszIn);
    }
    else if (strnicmp(p->pszIn, "dbnext ", 7) == 0)
    {
	GetNameKeyValue(pszCmdData, name, NULL, NULL);
	if (dbs_next(name, key, value) == DBS_SUCCESS)
	{
	    if (*key == '\0')
		ContextWriteString(g_pRightContext, p->pszIn);
	    else
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "dbresult dest=%s sock=%s result=key=%s value=%s", pszSrc, pszBfd, key, value);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "dbdelete ", 9) == 0)
    {
	GetNameKeyValue(pszCmdData, name, key, NULL);
	if (dbs_delete(name, key) == DBS_SUCCESS)
	{
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbresult dest=%s sock=%s result=DBS_SUCCESS", pszSrc, pszBfd);
	    ContextWriteString(g_pRightContext, pszStr);
	}
	else
	{
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
    }
    else
    {
	err_printf("unknown command '%s'", p->pszIn);
    }
}

bool GetIPString(char *pszHost, char *pszIPStr)
{
    unsigned int a, b, c, d;
    struct hostent *pH;
    
    pH = gethostbyname(pszHost);
    if (pH == NULL)
	return false;

    a = (unsigned char)(pH->h_addr_list[0][0]);
    b = (unsigned char)(pH->h_addr_list[0][1]);
    c = (unsigned char)(pH->h_addr_list[0][2]);
    d = (unsigned char)(pH->h_addr_list[0][3]);

    sprintf(pszIPStr, "%u.%u.%u.%u", a, b, c, d);

    return true;
}

void HandleLeftRead(MPD_Context *p)
{
    MPD_Context *pContext;
    char pszStr[MAX_CMD_LENGTH];
    char pszHost[MAX_HOST_LENGTH] = "";

    dbg_printf("LeftRead[%d]: '%s'\n", p->sock, p->pszIn);

    if (strnicmp(p->pszIn, "db", 2) == 0)
    {
	HandleDBCommandRead(p);
    }
    else if (strnicmp(p->pszIn, "launch ", 7) == 0)
    {
	pszHost[0] = '\0';
	GetStringOpt(p->pszIn, "h", pszHost);
	if ((stricmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	{
	    Launch(p->pszIn);
	}
	else
	{
	    bool bNoHost = pszHost[0] == '\0';
	    GetStringOpt(p->pszIn, "src", pszHost);
	    if (strcmp(pszHost, g_pszHost) == 0)
	    {
		if (bNoHost)
		{
		    // This needs to be handled by HandleConsoleRead
		    Launch(p->pszIn);
		}
		else
		{
		    if (GetStringOpt(p->pszIn, "try", pszStr))
		    {
			//dbg_printf("launch command went full circle without a match, discarding\n");
			char pszId[10];
			GetStringOpt(p->pszIn, "src", pszHost);
			GetStringOpt(p->pszIn, "id", pszId);
			_snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=invalid host", g_pszHost, pszHost, pszId);
			ContextWriteString(g_pRightContext, pszStr);
		    }
		    else
		    {
			pszHost[0] = '\0';
			GetStringOpt(p->pszIn, "h", pszHost);
			
			if (GetIPString(pszHost, pszHost))
			{
			    _snprintf(pszStr, MAX_CMD_LENGTH, "launch h=%s try=2 %s", pszHost, &p->pszIn[7]);
			    dbg_printf("trying launch again with ip string replacing the old hostname\n");
			    ContextWriteString(g_pRightContext, pszStr);
			}
			else
			{
			    char pszId[10];
			    GetStringOpt(p->pszIn, "src", pszHost);
			    GetStringOpt(p->pszIn, "id", pszId);
			    _snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=invalid host", g_pszHost, pszHost, pszId);
			    ContextWriteString(g_pRightContext, pszStr);
			}
		    }
		}
	    }
	    else
	    {
		dbg_printf("forwarding launch command\n");
		ContextWriteString(g_pRightContext, p->pszIn);
	    }
	}
    }
    else if (strnicmp(p->pszIn, "launched ", 9) == 0)