run.cpp

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

	pszExe += str;
    }

    pszExe.TrimRight();
    pszArgs.TrimLeft();
    pszArgs.TrimRight();
}

static void CmdLineToUnc(CString &pszApp)
{
    CString pszExe, pszArgs;
    SeparateCommand(pszApp, pszExe, pszArgs);
    ExeToUnc(pszExe);
    pszApp = pszExe;
    if (pszArgs.GetLength())
	pszApp += " " + pszArgs;
}

// Function name	: ParseLineIntoHostNode
// Description	    : 
// Return type		: HostNode* 
// Argument         : char * line
static HostNode* ParseLineIntoHostNode(char * line)
{
    char buffer[1024];
    char *pChar, *pChar2;
    HostNode *node = NULL;
    
    strncpy(buffer, line, 1024);
    buffer[1023] = '\0';
    pChar = buffer;
    
    // Advance over white space
    while (*pChar != '\0' && isspace(*pChar))
	pChar++;
    if (*pChar == '#' || *pChar == '\0')
	return NULL;
    
    // Trim trailing white space
    pChar2 = &buffer[strlen(buffer)-1];
    while (isspace(*pChar2) && (pChar >= pChar))
    {
	*pChar2 = '\0';
	pChar2--;
    }
    
    // If there is anything left on the line, consider it a host name
    if (strlen(pChar) > 0)
    {
	node = new HostNode;
	node->nSMPProcs = 1;
	node->next = NULL;
	node->exe[0] = '\0';
	
	// Copy the host name
	pChar2 = node->host;
	while (*pChar != '\0' && !isspace(*pChar))
	{
	    *pChar2 = *pChar;
	    pChar++;
	    pChar2++;
	}
	*pChar2 = '\0';
	
	// Advance over white space
	while (*pChar != '\0' && isspace(*pChar))
	    pChar++;
	// Get the number of SMP processes
	if (*pChar != '\0')
	{
	    node->nSMPProcs = atoi(pChar);
	    if (node->nSMPProcs < 1)
		node->nSMPProcs = 1;
	}
	// Advance over the number
	while (*pChar != '\0' && isdigit(*pChar))
	    pChar++;
	
	// Advance over white space
	while (*pChar != '\0' && isspace(*pChar))
	    pChar++;
	// Copy the executable
	if (*pChar != '\0')
	{
	    strncpy(node->exe, pChar, MAX_CMD_LENGTH);
	    node->exe[MAX_CMD_LENGTH-1] = '\0';
	    ExeToUnc(node->exe);
	}
    }
    
    return node;
}

// Function name	: ParseConfigFile
// Description	    : 
// Return type		: int 
int CGuiMPIRunView::ParseConfigFile()
{
    CString sArgs("");
    FILE *fin;
    char buffer[1024] = "";

    fin = fopen(m_ConfigFileName, "r");
    if (fin == NULL)
    {
	return PARSE_ERR_NO_FILE;
    }
    
    while (fgets(buffer, 1024, fin))
    {
	// Check for the name of the executable
	if (strnicmp(buffer, "exe ", 4) == 0)
	{
	    char *pChar = &buffer[4];
	    while (isspace(*pChar))
		pChar++;
	    m_app = pChar;
	    m_app.TrimRight();
	    ExeToUnc(m_app);
	    m_app = "\"" + m_app + "\"";
	}
	else
	{
	    // Check for program arguments
	    if (strnicmp(buffer, "args ", 5) == 0)
	    {
		sArgs = &buffer[5];
		sArgs.TrimLeft();
		sArgs.TrimRight();
	    }
	    else
	    {
		// Check for environment variables
		if (strnicmp(buffer, "env ", 4) == 0)
		{
		    m_CommonEnvironment = &buffer[4];
		    m_CommonEnvironment.TrimLeft();
		    m_CommonEnvironment.TrimRight();
		    if (m_CommonEnvironment.GetLength() > 0)
			m_bUseCommonEnvironment = true;
		}
		else
		{
		    // Check for hosts
		    if (strnicmp(buffer, "hosts", 5) == 0)
		    {
			m_nproc = 0;
			m_pHosts = NULL;
			HostNode *node, dummy;
			dummy.next = NULL;
			node = &dummy;
			while (fgets(buffer, 1024, fin))
			{
			    node->next = ParseLineIntoHostNode(buffer);
			    if (node->next != NULL)
			    {
				node = node->next;
				m_nproc = m_nproc + node->nSMPProcs;
			    }
			}
			m_pHosts = dummy.next;
			
			fclose(fin);

			if (sArgs.GetLength() > 0)
			    m_app = m_app + " " + sArgs;

			return PARSE_SUCCESS;
		    }
		}
	    }
	}
    }
    fclose(fin);
    if (sArgs.GetLength() > 0)
	m_app = m_app + " " + sArgs;

    return PARSE_SUCCESS;
}

static MapDriveNode *MakeMapFromString(CString str)
{
    if (str.GetLength() > 7 && str[1] == ':')
    {
	MapDriveNode *pNode = new MapDriveNode;
	pNode->cDrive = str[0];
	strcpy(pNode->pszShare, str.Right(str.GetLength()-2));
	pNode->pNext = NULL;
	return pNode;
    }
    return NULL;
}

void RunJob(CGuiMPIRunView *pDlg)
{
    int i;
    int iproc = 0;
    char pszJobID[100];
    char pszEnv[MAX_CMD_LENGTH] = "";
    char pszDir[MAX_PATH] = ".";
    int nShmLow, nShmHigh;
    DWORD dwThreadID;
    char pBuffer[MAX_CMD_LENGTH];
    char cMapDrive;
    CString pszMapShare;
    CString sAppOriginal;
    CString sMapping = "";

    try{
    sAppOriginal = pDlg->m_app;
    CmdLineToUnc(pDlg->m_app);
    if (pDlg->m_bUseSlaveProcess)
	CmdLineToUnc(pDlg->m_SlaveProcess);

    if (!pDlg->m_bUseConfigFile)
    {
	// Make a list of hosts to launch on based on m_nproc and m_host_list
	pDlg->GetHosts();
    }

    /*
    if (pDlg->m_bUseMapping)
    {
	// Build the mapping nodes here
	CString str, sub;
	MapDriveNode *pNode;

	str = pDlg->m_Mappings;
	int index = 0, n;
	n = str.Find(';', index);
	while (n != -1)
	{
	    sub = str.Mid(index, n-index);
	    pNode = MakeMapFromString(sub);
	    if (pNode != NULL)
	    {
		pNode->pNext = pDlg->m_pDriveMapList;
		pDlg->m_pDriveMapList = pNode;
	    }
	    n++; // skip over the ;
	    index = n;
	    n = str.Find(';', index);
	}
	sub = str.Mid(index);
	pNode = MakeMapFromString(sub);
	if (pNode != NULL)
	{
	    pNode->pNext = pDlg->m_pDriveMapList;
	    pDlg->m_pDriveMapList = pNode;
	}
    }
    */

    // Create a job id string
    CreateJobID(pszJobID);
    
    // Set the environment variables common to all processes
    if (pDlg->m_bNoMPI)
    {
	if (pDlg->m_bUseCommonEnvironment && (pDlg->m_CommonEnvironment.GetLength() > 0))
	{
	    if (_snprintf(pszEnv, MAX_CMD_LENGTH, "%s", pDlg->m_CommonEnvironment) < 0)
	    {
		// environment variables truncated
		pszEnv[MAX_CMD_LENGTH-1] = '\0';
	    }
	}
	else
	    pszEnv[0] = '\0';
    }
    else
    {
	if (pDlg->m_bUseCommonEnvironment && (pDlg->m_CommonEnvironment.GetLength() > 0))
	{
	    if (_snprintf(pszEnv, MAX_CMD_LENGTH, "%s|MPICH_JOBID=%s|MPICH_NPROC=%d|MPICH_ROOTHOST=%s",
		pDlg->m_CommonEnvironment, pszJobID, pDlg->m_nproc, pDlg->m_pHosts->host) < 0)
	    {
		// environment variables truncated
		pszEnv[MAX_CMD_LENGTH-1] = '\0';
	    }
	}
	else
	{
	    if (_snprintf(pszEnv, MAX_CMD_LENGTH, "MPICH_JOBID=%s|MPICH_NPROC=%d|MPICH_ROOTHOST=%s",
		pszJobID, pDlg->m_nproc, pDlg->m_pHosts->host) < 0)
	    {
		// environment variables truncated
		pszEnv[MAX_CMD_LENGTH-1] = '\0';
	    }
	}
    }

    // Get the directory from the executable or use the user selected working directory
    if (pDlg->m_bUseWorkingDirectory)
    {
	strcpy(pszDir, pDlg->m_WorkingDirectory);
    }
    else
    {
	if (sAppOriginal[0] == '\\' && sAppOriginal[1] == '\\')
	{
	    GetCurrentDirectory(MAX_PATH, pszDir);
	}
	else
	{
	    char pszTempExe[MAX_CMD_LENGTH] = ".", *namepart;
	    if (GetFullPathName(sAppOriginal, MAX_CMD_LENGTH, pszTempExe, &namepart) > 0)
	    {
		if (namepart - pszTempExe < MAX_CMD_LENGTH)
		{
		    strncpy(pszDir, pszTempExe, namepart - pszTempExe);
		    pszDir[namepart - pszTempExe] = '\0';
		}
	    }
	    else
	    {
		CString str = sAppOriginal.Left(pDlg->m_app.ReverseFind('\\'));
		if (GetFullPathName(str, MAX_CMD_LENGTH, pszTempExe, &namepart) > 0)
		{
		    if (namepart - pszTempExe < MAX_CMD_LENGTH)
		    {
			strncpy(pszDir, pszTempExe, namepart - pszTempExe);
			pszDir[namepart - pszTempExe] = '\0';
		    }
		}
	    }
	}
    }
    if (NeedToMap(pszDir, &cMapDrive, pszMapShare))
    {
	if (pDlg->m_bUseMapping)
	    sMapping.Format(" m='%c:%s;%s'", cMapDrive, pszMapShare, pDlg->m_Mappings);
	else
	    sMapping.Format(" m='%c:%s'", cMapDrive, pszMapShare);
	/*
	MapDriveNode *pNode = new MapDriveNode;
	pNode->cDrive = cMapDrive;
	strcpy(pNode->pszShare, pszMapShare);
	pNode->pNext = pDlg->m_pDriveMapList;
	pDlg->m_pDriveMapList = pNode;
	*/
    }
    else
    {
	if (pDlg->m_bUseMapping)
	    sMapping.Format(" m='%s'", pDlg->m_Mappings);
    }

    // Allocate an array to hold handles to the LaunchProcess threads
    pDlg->m_nNumProcessThreads = 0;
    pDlg->m_pProcessThread = new HANDLE[pDlg->m_nproc];
    pDlg->m_pProcessSocket = new SOCKET[pDlg->m_nproc];
    pDlg->m_pProcessLaunchId = new int[pDlg->m_nproc];
    pDlg->m_pLaunchIdToRank = new int[pDlg->m_nproc];
    pDlg->m_nNumProcessSockets = 0;
    pDlg->m_pForwardHost = new ForwardHostStruct[pDlg->m_nproc];
    for (i=0; i<pDlg->m_nproc; i++)
	pDlg->m_pForwardHost[i].nPort = 0;
    
    // Start the IO redirection thread
    RedirectIOArg *pArg = new RedirectIOArg;
    pArg->hReadyEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
    pArg->pDlg = pDlg;
    pDlg->m_hRedirectIOListenThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RedirectIOThread, pArg, 0, &dwThreadID);
    if (pDlg->m_hRedirectIOListenThread)
    {
	if (WaitForSingleObject(pArg->hReadyEvent, 10000) != WAIT_OBJECT_0)
	{
	    MessageBox(NULL, "RedirectIOThread failed to initialize", "Error", MB_OK);
	    delete pArg;
	    delete pDlg->m_pProcessThread;
	    delete pDlg->m_pProcessSocket;
	    delete pDlg->m_pProcessLaunchId;
	    delete pDlg->m_pLaunchIdToRank;
	    delete pDlg->m_pForwardHost;
	    pDlg->m_pProcessThread = NULL;
	    pDlg->m_pProcessSocket = NULL;
	    pDlg->m_nNumProcessSockets = 0;
	    pDlg->m_pProcessLaunchId = NULL;
	    pDlg->m_pLaunchIdToRank = NULL;
	    pDlg->m_pForwardHost = NULL;
	    return;
	}
    }
    else
    {
	char str[256];
	sprintf(str, "Unable to create RedirectIOThread, error %d\n", GetLastError());
	MessageBox(NULL, str, "Error", MB_OK);
	delete pArg;
	delete pDlg->m_pProcessThread;
	delete pDlg->m_pProcessSocket;
	delete pDlg->m_pProcessLaunchId;
	delete pDlg->m_pLaunchIdToRank;
	delete pDlg->m_pForwardHost;
	pDlg->m_pProcessThread = NULL;
	pDlg->m_pProcessSocket = NULL;
	pDlg->m_nNumProcessSockets = 0;
	pDlg->m_pProcessLaunchId = NULL;
	pDlg->m_pLaunchIdToRank = NULL;
	pDlg->m_pForwardHost = NULL;
	return;
    }
    CloseHandle(pArg->hReadyEvent);
    delete pArg;

    // Copy the io redirection thread stuff into the first forwarder entry
    strncpy(pDlg->m_pForwardHost[0].pszHost, pDlg->m_pszIOHost, MAX_HOST_LENGTH);
    pDlg->m_pForwardHost[0].pszHost[MAX_HOST_LENGTH-1] = '\0';
    pDlg->m_pForwardHost[0].nPort = pDlg->m_nIOPort;

    // Launch the processes
    while (pDlg->m_pHosts)
    {
	nShmLow = iproc;
	nShmHigh = iproc + pDlg->m_pHosts->nSMPProcs - 1;
	for (int i = 0; i<pDlg->m_pHosts->nSMPProcs; i++)
	{
	    MPIRunLaunchProcessArg *arg = new MPIRunLaunchProcessArg;
	    if (sMapping.GetLength() > 0)
		strcpy(arg->pszMap, sMapping);
	    else
		arg->pszMap[0] = '\0';
	    arg->bUseDebugFlag = pDlg->m_bCatch;
	    arg->pDlg = pDlg;
	    arg->n = pDlg->m_nproc;
	    sprintf(arg->pszIOHostPort, "%s:%d", pDlg->m_pszIOHost, pDlg->m_nIOPort);
	    strcpy(arg->pszPassPhrase, pDlg->m_Phrase);
	    arg->i = iproc;
	    arg->bLogon = pDlg->m_bLogon;
	    if (pDlg->m_bLogon)
	    {
		strcpy(arg->pszAccount, pDlg->m_Account);
		strcpy(arg->pszPassword, pDlg->m_Password);
	    }
	    if (strlen(pDlg->m_pHosts->exe) > 0)
	    {
		strncpy(arg->pszCmdLine, pDlg->m_pHosts->exe, MAX_CMD_LENGTH);
		arg->pszCmdLine[MAX_CMD_LENGTH-1] = '\0';
	    }
	    else
	    {