launch.cpp

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

#include "GetStringOpt.h"
#include "mpdimpl.h"
#include "mpdutil.h"
#include <winsock2.h>
#include <windows.h>
#include <stdio.h>
#include "Translate_Error.h"
#include "safe_terminate_process.h"

long g_nNumProcsRunning = 0;
HANDLE g_hProcessStructMutex = NULL;

struct LaunchThreadStruct
{
    LaunchThreadStruct();
    void Print();

    char pszHost[MAX_HOST_LENGTH];
    char pszSrcHost[MAX_HOST_LENGTH];
    char pszSrcId[10];
    char pszEnv[MAX_CMD_LENGTH];
    char pszMap[MAX_CMD_LENGTH];
    char pszDir[MAX_PATH];
    char pszCmd[MAX_CMD_LENGTH];
    char pszAccount[40];
    char pszPassword[300];
    char pszStdin[MAX_HOST_LENGTH];
    char pszStdout[MAX_HOST_LENGTH];
    char pszStderr[MAX_HOST_LENGTH];
    bool bMergeOutErr;
    bool bUseDebugFlag;
    int priorityClass;
    int priority;
    bool bAttachToWorkstation;

    int nPid;
    int nKRank;
    char pszError[MAX_CMD_LENGTH];
    int nExitCode;
    HANDLE hProcess;
    HANDLE hThread;

    LaunchThreadStruct *pNext;
};

LaunchThreadStruct::LaunchThreadStruct()
{
    pszHost[0] = '\0';
    pszSrcHost[0] = '\0';
    pszSrcId[0] = '\0';
    pszEnv[0] = '\0';
    pszMap[0] = '\0';
    pszDir[0] = '\0';
    pszCmd[0] = '\0';
    pszAccount[0] = '\0';
    pszPassword[0] = '\0';
    pszStdin[0] = '\0';
    pszStdout[0] = '\0';
    pszStderr[0] = '\0';
    bMergeOutErr = false;
    bUseDebugFlag = false;
    priorityClass = BELOW_NORMAL_PRIORITY_CLASS;
    priority = THREAD_PRIORITY_NORMAL;
    bAttachToWorkstation = false;

    nPid = -1;
    nKRank = 0;
    pszError[0] = '\0';
    nExitCode = -1;
    hProcess = NULL;
    hThread = NULL;
    
    pNext = NULL;
}

void LaunchThreadStruct::Print()
{
    dbg_printf("LAUNCH:\n");
    dbg_printf(" user: %s\n", pszAccount);
    dbg_printf(" %s(%s) -> %s %s\n", pszSrcHost, pszSrcId, pszHost, pszCmd);
    if (pszDir[0] != '\0')
    {
	dbg_printf(" dir: ");
	int n = strlen(pszDir);
	if (n > 70)
	{
	    char pszTemp[71];
	    char *pszCur = pszDir;
	    bool bFirst = true;
	    while (n > 0)
	    {
		strncpy(pszTemp, pszCur, 70);
		pszTemp[70] = '\0';
		if (bFirst)
		{
		    printf("%s\n", pszTemp);
		    bFirst = false;
		}
		else
		    printf("      %s\n", pszTemp);
		pszCur += 70;
		n -= 70;
	    }
	}
	else
	    printf("%s\n", pszDir);
    }
    if (pszEnv[0] != '\0')
    {
	char pszEnv2[MAX_CMD_LENGTH];
	char pszCheck[100];
	char *token;
	int i,n;
	strncpy(pszEnv2, pszEnv, MAX_CMD_LENGTH);
	pszEnv2[MAX_CMD_LENGTH-1] = '\0';

	token = strstr(pszEnv2, "PMI_PWD=");
	if (token != NULL)
	{
	    strncpy(pszCheck, &token[8], 100);
	    pszCheck[99] = '\0';
	    token = strtok(pszCheck, " '|\n");
	    n = strlen(pszCheck);
	    token = strstr(pszEnv2, "PMI_PWD=");
	    token = &token[8];
	    if (n > 0)
	    {
		if (pszCheck[n-1] == '\r' || pszCheck[n-1] == '\n')
		    n--;
		if (pszCheck[n-1] == '\r' || pszCheck[n-1] == '\n')
		    n--;
		for (i=0; i<n; i++)
		    token[i] = '*';
	    }
	}

	printf(" env: ");
	n = strlen(pszEnv2);
	if (n > 70)
	{
	    char pszTemp[71];
	    char *pszCur = pszEnv2;
	    bool bFirst = true;
	    while (n > 0)
	    {
		strncpy(pszTemp, pszCur, 70);
		pszTemp[70] = '\0';
		if (bFirst)
		{
		    printf("%s\n", pszTemp);
		    bFirst = false;
		}
		else
		    printf("      %s\n", pszTemp);
		pszCur += 70;
		n -= 70;
	    }
	}
	else
	    printf("%s\n", pszEnv2);
    }
    if (pszMap[0] != '\0')
	printf(" map = %s\n", pszMap);
    printf(" stdin|out|err: %s|%s|%s\n", pszStdin, pszStdout, pszStderr);
    printf(" krank: %d\n", nKRank);
    //printf("\n");
    fflush(stdout);
}

LaunchThreadStruct *g_pProcessList = NULL;

bool snprintf_update(char *&pszStr, int &length, char *pszFormat, ...)
{
    va_list list;
    int n;

    va_start(list, pszFormat);
    n = _vsnprintf(pszStr, length, pszFormat, list);
    va_end(list);

    if (n < 0)
    {
	pszStr[length-1] = '\0';
	length = 0;
	return false;
    }

    pszStr = &pszStr[n];
    length = length - n;

    return true;
}

static void ProcessToString(LaunchThreadStruct *p, char *pszStr, int length)
{
    if (!snprintf_update(pszStr, length, "PROCESS:\n"))
	return;
    if (p->pszAccount[0] != '\0')
    {
	if (!snprintf_update(pszStr, length, " user: %s\n", p->pszAccount))
	    return;
    }
    else
    {
	if (!snprintf_update(pszStr, length, " user: <single user mode>\n"))
	    return;
    }
    if (!snprintf_update(pszStr, length, " %s(%s) -> %s %s\n", p->pszSrcHost, p->pszSrcId, p->pszHost, p->pszCmd))
	return;
    if (p->pszDir[0] != '\0')
    {
	if (!snprintf_update(pszStr, length, " dir: "))
	    return;
	int n = strlen(p->pszDir);
	if (n > 70)
	{
	    char pszTemp[71];
	    char *pszCur = p->pszDir;
	    bool bFirst = true;
	    while (n > 0)
	    {
		strncpy(pszTemp, pszCur, 70);
		pszTemp[70] = '\0';
		if (bFirst)
		{
		    if (!snprintf_update(pszStr, length, "%s\n", pszTemp))
			return;
		    bFirst = false;
		}
		else
		{
		    if (!snprintf_update(pszStr, length, "      %s\n", pszTemp))
			return;
		}
		pszCur += 70;
		n -= 70;
	    }
	}
	else
	{
	    if (!snprintf_update(pszStr, length, "%s\n", p->pszDir))
		return;
	}
    }
    if (p->pszEnv[0] != '\0')
    {
	char pszEnv2[MAX_CMD_LENGTH];
	char pszCheck[100];
	char *token;
	int i,n;
	strncpy(pszEnv2, p->pszEnv, MAX_CMD_LENGTH);
	pszEnv2[MAX_CMD_LENGTH-1] = '\0';

	token = strstr(pszEnv2, "PMI_PWD=");
	if (token != NULL)
	{
	    strncpy(pszCheck, &token[8], 100);
	    pszCheck[99] = '\0';
	    token = strtok(pszCheck, " '|\n");
	    n = strlen(pszCheck);
	    token = strstr(pszEnv2, "PMI_PWD=");
	    token = &token[8];
	    if (n > 0)
	    {
		if (pszCheck[n-1] == '\r' || pszCheck[n-1] == '\n')
		    n--;
		if (pszCheck[n-1] == '\r' || pszCheck[n-1] == '\n')
		    n--;
		for (i=0; i<n; i++)
		    token[i] = '*';
	    }
	}

	if (!snprintf_update(pszStr, length, " env: "))
	    return;
	n = strlen(pszEnv2);
	if (n > 70)
	{
	    char pszTemp[71];
	    char *pszCur = pszEnv2;
	    bool bFirst = true;
	    while (n > 0)
	    {
		strncpy(pszTemp, pszCur, 70);
		pszTemp[70] = '\0';
		if (bFirst)
		{
		    if (!snprintf_update(pszStr, length, "%s\n", pszTemp))
			return;
		    bFirst = false;
		}
		else
		{
		    if (!snprintf_update(pszStr, length, "      %s\n", pszTemp))
			return;
		}
		pszCur += 70;
		n -= 70;
	    }
	}
	else
	{
	    if (!snprintf_update(pszStr, length, "%s\n", pszEnv2))
		return;
	}
    }
    if (!snprintf_update(pszStr, length, " stdin|out|err: %s|%s|%s\n", p->pszStdin, p->pszStdout, p->pszStderr))
	return;
    if (!snprintf_update(pszStr, length, " krank: %d\n", p->nKRank))
	return;
}

void statProcessList(char *pszOutput, int length)
{
    LaunchThreadStruct *p;
    int nBytesAvailable = length;

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

    // lock the process list while using it
    WaitForSingleObject(g_hProcessStructMutex, INFINITE);

    if (g_pProcessList == NULL)
    {
	ReleaseMutex(g_hProcessStructMutex);
	return;
    }

    p = g_pProcessList;
    while (p)
    {
	ProcessToString(p, pszOutput, length);
	length = length - strlen(pszOutput);
	pszOutput = &pszOutput[strlen(pszOutput)];
	p = p->pNext;
    }
    ReleaseMutex(g_hProcessStructMutex);
}

void RemoveProcessStruct(LaunchThreadStruct *p)
{
    WaitForSingleObject(g_hProcessStructMutex, INFINITE);

    LaunchThreadStruct *pTrailer = g_pProcessList;

    // Remove p from the list
    if (p == NULL)
    {
	ReleaseMutex(g_hProcessStructMutex);
	return;
    }

    if (p == g_pProcessList)
	g_pProcessList = g_pProcessList->pNext;
    else
    {
	while (pTrailer && pTrailer->pNext != p)
	    pTrailer = pTrailer->pNext;
	if (pTrailer)
	    pTrailer->pNext = p->pNext;
    }

    // Close any open handles
    if (p->hProcess != NULL)
	CloseHandle(p->hProcess);
    if (p->hThread != NULL)
	CloseHandle(p->hThread);

    UnmapUserDrives(p->pszMap);

    //dbg_printf("removing ProcessStruct[%d]\n", p->nPid);
    // free the structure
    delete p;

    ReleaseMutex(g_hProcessStructMutex);
}

void LaunchThread(LaunchThreadStruct *pArg)
{
    DWORD dwExitCode;
    char pszStr[MAX_CMD_LENGTH];
    char pszError[MAX_PATH];
    bool bProcessAborted = false;
    SYSTEMTIME stime;
    char timestamp[256];
    HANDLE hIn, hOut, hErr;
    int nError;

    pArg->Print();

    if (strlen(pArg->pszEnv))
    {
	sprintf(pszStr, "|MPD_ID=%s", pArg->pszSrcId);
	strcat(pArg->pszEnv, pszStr);
    }
    else
    {
	sprintf(pArg->pszEnv, "MPD_ID=%s", pArg->pszSrcId);
    }
    pszStr[0] = '\0';
    if (g_bSingleUser)
    {