launch.cpp

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

	if (!MapUserDrives(pArg->pszMap, pArg->pszAccount, pArg->pszPassword, pszStr))
	{
	}
	pArg->hProcess = LaunchProcess(pArg->pszCmd, pArg->pszEnv, pArg->pszDir, 
	    //BELOW_NORMAL_PRIORITY_CLASS, THREAD_PRIORITY_NORMAL,
	    pArg->priorityClass, pArg->priority,
	    &hIn, &hOut, &hErr, &pArg->nPid, &nError, pszStr, pArg->bUseDebugFlag);
    }
    else
    {
	if (pArg->pszAccount[0] == '\0')
	{
	    if (g_bMPDUserCapable)
	    {
		if (g_bUseMPDUser)
		{
		    _snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=LaunchProcess failed, invalid mpd user for anonymous launch.", g_pszHost, pArg->pszSrcHost, pArg->pszSrcId);
		}
		else
		{
		    _snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=LaunchProcess failed, anonymous launch not enabled on '%s'.", g_pszHost, pArg->pszSrcHost, pArg->pszSrcId, g_pszHost);
		}
	    }
	    else
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=LaunchProcess failed, anonymous launch request attempted on node without that capability enabled.", g_pszHost, pArg->pszSrcHost, pArg->pszSrcId);
	    }
	    ContextWriteString(g_pRightContext, pszStr);
	    RemoveProcessStruct(pArg);
	    InterlockedDecrement(&g_nNumProcsRunning);
	    return;
	}
	pArg->hProcess = LaunchProcessLogon(pArg->pszAccount, pArg->pszPassword, 
				pArg->pszCmd, pArg->pszEnv, pArg->pszMap, pArg->pszDir, 
				//BELOW_NORMAL_PRIORITY_CLASS, THREAD_PRIORITY_NORMAL,
				pArg->priorityClass, pArg->priority,
				&hIn, &hOut, &hErr, &pArg->nPid, &nError, pszStr, pArg->bUseDebugFlag);
    }
    if (pArg->hProcess == INVALID_HANDLE_VALUE)
    {
	Translate_Error(nError, pszError, pszStr);
	_snprintf(pszStr, MAX_CMD_LENGTH, "launched src=%s dest=%s id=%s error=LaunchProcess failed, %s", g_pszHost, pArg->pszSrcHost, pArg->pszSrcId, pszError);
	ContextWriteString(g_pRightContext, pszStr);
	RemoveProcessStruct(pArg);
	InterlockedDecrement(&g_nNumProcsRunning);
	return;
    }

    _snprintf(pszStr, MAX_CMD_LENGTH, "launched pid=%d src=%s dest=%s id=%s", pArg->nPid, g_pszHost, pArg->pszSrcHost, pArg->pszSrcId);
    ContextWriteString(g_pRightContext, pszStr);

    if (!ConnectAndRedirectInput(hIn, pArg->pszStdin, pArg->hProcess, pArg->nPid, pArg->nKRank))
    {
	if (!SafeTerminateProcess(pArg->hProcess, 1000001))
	{
	    if (GetLastError() != ERROR_PROCESS_ABORTED)
		TerminateProcess(pArg->hProcess, 1000006);
	}
    }
    else 
    {
	if (pArg->bMergeOutErr)
	{
	    if (!ConnectAndRedirect2Outputs(hOut, hErr, pArg->pszStdout, pArg->hProcess, pArg->nPid, pArg->nKRank))
	    {
		if (!SafeTerminateProcess(pArg->hProcess, 1000002))
		{
		    if (GetLastError() != ERROR_PROCESS_ABORTED)
			TerminateProcess(pArg->hProcess, 1000007);
		}
	    }
	}
	else
	{
	    if (!ConnectAndRedirectOutput(hOut, pArg->pszStdout, pArg->hProcess, pArg->nPid, pArg->nKRank, 1))
	    {
		if (!SafeTerminateProcess(pArg->hProcess, 1000003))
		{
		    if (GetLastError() != ERROR_PROCESS_ABORTED)
			TerminateProcess(pArg->hProcess, 1000008);
		}
	    }
	    else if (!ConnectAndRedirectOutput(hErr, pArg->pszStderr, pArg->hProcess, pArg->nPid, pArg->nKRank, 2))
	    {
		if (!SafeTerminateProcess(pArg->hProcess, 1000004))
		{
		    if (GetLastError() != ERROR_PROCESS_ABORTED)
			TerminateProcess(pArg->hProcess, 1000009);
		}
	    }
	}
    }

    if (pArg->bUseDebugFlag)
    {
	DebugWaitForProcess(bProcessAborted, pszError);
    }

    WaitForSingleObject(pArg->hProcess, INFINITE);
    GetLocalTime(&stime);
    dwExitCode = 123456789;
    GetExitCodeProcess(pArg->hProcess, &dwExitCode);
    pArg->nExitCode = dwExitCode;
    CloseHandle(pArg->hProcess);
    pArg->hProcess = NULL;

#ifdef FOO
    if (dwExitCode == ERROR_WAIT_NO_CHILDREN)
    {
	sprintf(pszError, "unexpected process termination.");
	bProcessAborted = true;
    }
#endif

    sprintf(timestamp, "%d.%d.%d.%dh.%dm.%ds.%dms", 
	stime.wYear, stime.wMonth, stime.wDay, 
	stime.wHour, stime.wMinute, stime.wSecond, stime.wMilliseconds);
    if (bProcessAborted)
    {
	_snprintf(pszStr, MAX_CMD_LENGTH, "exitcode code=%d src=%s dest=%s id=%s time=%s error=%s", 
	    pArg->nExitCode, g_pszHost, pArg->pszSrcHost, pArg->pszSrcId, timestamp, pszError);
    }
    else
    {
	_snprintf(pszStr, MAX_CMD_LENGTH, "exitcode code=%d src=%s dest=%s id=%s time=%s", 
	    pArg->nExitCode, g_pszHost, pArg->pszSrcHost, pArg->pszSrcId, timestamp);
    }
    dbg_printf("...process %d exited, sending <%s>\n", pArg->nKRank, pszStr);
    ContextWriteString(g_pRightContext, pszStr);

    RemoveProcessStruct(pArg);
    InterlockedDecrement(&g_nNumProcsRunning);
}

void ShutdownAllProcesses()
{
    //DWORD dwExitCode;
    WaitForSingleObject(g_hProcessStructMutex, INFINITE);
    LaunchThreadStruct *p = g_pProcessList;
    while (p)
    {
	if (p->hProcess)
	{
	    if (!SafeTerminateProcess(p->hProcess, 1000005))
	    {
		if (GetLastError() != ERROR_PROCESS_ABORTED)
		{
		    if (!TerminateProcess(p->hProcess, 1000006))
		    {
			InterlockedDecrement(&g_nNumProcsRunning);
		    }
		}
	    }
	    /*
	    if (GetExitCodeProcess(p->hProcess, &dwExitCode))
	    {
		if (dwExitCode == STILL_ACTIVE)
		{
		    if (!TerminateProcess(p->hProcess, 0))
		    {
			err_printf("TerminateProcess failed for process %d, %d, error %d\n", p->hProcess, p->nPid, GetLastError());
			// If I can't stop a process for some reason,
			// decrement its value so this function doesn't hang
			InterlockedDecrement(&g_nNumProcsRunning);
		    }
		}
	    }
	    */
	}
	p = p->pNext;
    }
    ReleaseMutex(g_hProcessStructMutex);

    // Wait for all the threads to clean up the terminated processes
    while (g_nNumProcsRunning > 0)
	Sleep(250);
}

void MPD_KillProcess(int nPid)
{
    //DWORD dwExitCode;
    WaitForSingleObject(g_hProcessStructMutex, INFINITE);
    LaunchThreadStruct *p = g_pProcessList;
    while (p)
    {
	if (p->nPid == nPid)
	{
	    //dbg_printf("MPD_KillProcess found pid %d\n", nPid);
	    if (p->hProcess && (p->hProcess != INVALID_HANDLE_VALUE))
	    {
		if (!SafeTerminateProcess(p->hProcess, 987654321))
		{
		    if (GetLastError() != ERROR_PROCESS_ABORTED)
		    {
			if (p->hProcess == NULL)
			{
			    // If the process handle is lost for some reason,
			    // decrement its value so this function doesn't hang
			    InterlockedDecrement(&g_nNumProcsRunning);
			}
			else
			{
			    if (!TerminateProcess(p->hProcess, 123456789))
			    {
				err_printf("TerminateProcess failed for process - handle(0x%p), pid(%d), error %d\n", p->hProcess, p->nPid, GetLastError());
				// If I can't stop a process for some reason,
				// decrement its value so this function doesn't hang
				InterlockedDecrement(&g_nNumProcsRunning);
			    }
			}
		    }
		}
		/*
		//dbg_printf("MPD_KillProcess found valid hProcess 0x%x\n", p->hProcess);
		if (GetExitCodeProcess(p->hProcess, &dwExitCode))
		{
		    if (dwExitCode == STILL_ACTIVE)
		    {
			//dbg_printf("MPD_KillProcess - terminating process\n");
			if (!TerminateProcess(p->hProcess, 0))
			{
			    err_printf("TerminateProcess failed for process %d, %d, error %d\n", p->hProcess, p->nPid, GetLastError());
			    // If I can't stop a process for some reason,
			    // decrement its value so this function doesn't hang
			    InterlockedDecrement(&g_nNumProcsRunning);
			    // Should I also remove the LaunchThreadStruct p?
			    // If there are lots of failed process terminations, this will lead to wasted memory allocation.
			}
		    }
		}
		else
		{
		    err_printf("MPD_KillProcess - GetExitCodeProcess failed, error %d\n", GetLastError());
		}
		*/
	    }
	    ReleaseMutex(g_hProcessStructMutex);
	    return;
	}
	p = p->pNext;
    }
    ReleaseMutex(g_hProcessStructMutex);
}

void Launch(char *pszStr)
{
    char sTemp[10];
    HANDLE hTemp;
    LaunchThreadStruct *pArg = new LaunchThreadStruct;

    if (GetStringOpt(pszStr, "g", sTemp))
	pArg->bUseDebugFlag = (stricmp(sTemp, "yes") == 0);
    if (GetStringOpt(pszStr, "k", sTemp))
	pArg->nKRank = atoi(sTemp);
    else
	pArg->nKRank = 0;
    if (!GetStringOpt(pszStr, "h", pArg->pszHost))
	strncpy(pArg->pszHost, g_pszHost, MAX_HOST_LENGTH);
    GetStringOpt(pszStr, "src", pArg->pszSrcHost);
    GetStringOpt(pszStr, "id", pArg->pszSrcId);
    GetStringOpt(pszStr, "e", pArg->pszEnv);
    GetStringOpt(pszStr, "m", pArg->pszMap);
    GetStringOpt(pszStr, "d", pArg->pszDir);
    GetStringOpt(pszStr, "c", pArg->pszCmd);
    if (GetStringOpt(pszStr, "a", pArg->pszAccount))
    {
	GetStringOpt(pszStr, "p", pArg->pszPassword);
	DecodePassword(pArg->pszPassword);
    }
    else
    {
	if (g_bMPDUserCapable && g_bUseMPDUser)
	{
	    strcpy(pArg->pszAccount, g_pszMPDUserAccount);
	    strcpy(pArg->pszPassword, g_pszMPDUserPassword);
	}
	else
	{
	    pArg->pszAccount[0] = '\0';
	    pArg->pszPassword[0] = '\0';
	}
    }
    GetStringOpt(pszStr, "0", pArg->pszStdin);
    GetStringOpt(pszStr, "1", pArg->pszStdout);
    GetStringOpt(pszStr, "2", pArg->pszStderr);
    if (GetStringOpt(pszStr, "12", pszStr))
    {
	strncpy(pArg->pszStdout, pszStr, MAX_HOST_LENGTH);
	strncpy(pArg->pszStderr, pszStr, MAX_HOST_LENGTH);
	pArg->bMergeOutErr = true;
    }
    if (GetStringOpt(pszStr, "012", pszStr))
    {
	strncpy(pArg->pszStdin, pszStr, MAX_HOST_LENGTH);
	strncpy(pArg->pszStdout, pszStr, MAX_HOST_LENGTH);
	strncpy(pArg->pszStderr, pszStr, MAX_HOST_LENGTH);
	pArg->bMergeOutErr = true;
    }
    if (GetStringOpt(pszStr, "r", pszStr))
    {
	int c,p;
	char *token;
	token = strtok(pszStr, ":");
	if (token)
	{
	    c = atoi(token);
	    switch (c)
	    {
	    case 0:
		pArg->priorityClass = IDLE_PRIORITY_CLASS;
		break;
	    case 1:
		pArg->priorityClass = BELOW_NORMAL_PRIORITY_CLASS;
		break;
	    case 2:
		pArg->priorityClass = NORMAL_PRIORITY_CLASS;
		break;
	    case 3:
		pArg->priorityClass = ABOVE_NORMAL_PRIORITY_CLASS;
		break;
	    case 4:
		pArg->priorityClass = HIGH_PRIORITY_CLASS;
		break;
	    default:
		pArg->priorityClass = BELOW_NORMAL_PRIORITY_CLASS;
		break;
	    }
	    token = strtok(NULL, " \n");
	    if (token != NULL)
	    {
		p = atoi(token);
		switch (p)
		{
		case 0:
		    pArg->priority = THREAD_PRIORITY_IDLE;
		    break;
		case 1:
		    pArg->priority = THREAD_PRIORITY_LOWEST;
		    break;
		case 2:
		    pArg->priority = THREAD_PRIORITY_BELOW_NORMAL;
		    break;
		case 3:
		    pArg->priority = THREAD_PRIORITY_NORMAL;
		    break;
		case 4:
		    pArg->priority = THREAD_PRIORITY_ABOVE_NORMAL;
		    break;
		case 5:
		    pArg->priority = THREAD_PRIORITY_HIGHEST;
		    break;
		default:
		    pArg->priority = THREAD_PRIORITY_NORMAL;
		    break;
		}
	    }
	}
    }

    WaitForSingleObject(g_hProcessStructMutex, INFINITE);
    if (!g_pProcessList)
    {
	g_pProcessList = pArg;
    }
    else
    {
	pArg->pNext = g_pProcessList;
	g_pProcessList = pArg;
    }
    ReleaseMutex(g_hProcessStructMutex);

    InterlockedIncrement(&g_nNumProcsRunning);

    DWORD dwThreadID;
    hTemp = pArg->hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)LaunchThread, pArg, 0, &dwThreadID);
    if (hTemp == NULL)
    {
	err_printf("Launch: CreateThread failed, error %d\n", GetLastError());
	InterlockedDecrement(&g_nNumProcsRunning);
    }

    return;
}

void ConcatenateProcessesToString(char *pszStr)
{
    char pszLine[4096];
    WaitForSingleObject(g_hProcessStructMutex, INFINITE);
    LaunchThreadStruct *p = g_pProcessList;
    if (p)
    {
	_snprintf(pszLine, 4096, "%s:\n", g_pszHost);
	strncat(pszStr, pszLine, MAX_CMD_LENGTH - 1 - strlen(pszStr));
    }
    while (p)
    {
	_snprintf(pszLine, 4096, "%04d : %s\n", p->nPid, p->pszCmd);
	strncat(pszStr, pszLine, MAX_CMD_LENGTH - 1 - strlen(pszStr));
	p = p->pNext;
    }
    ReleaseMutex(g_hProcessStructMutex);
}