run.cpp

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

		strncpy(arg->pszCmdLine, pDlg->m_app, MAX_CMD_LENGTH);
		arg->pszCmdLine[MAX_CMD_LENGTH-1] = '\0';
	    }
	    strcpy(arg->pszDir, pszDir);
	    if (strlen(pszEnv) >= MAX_CMD_LENGTH)
	    {
		// environment variables truncated
	    }
	    strncpy(arg->pszEnv, pszEnv, MAX_CMD_LENGTH);
	    arg->pszEnv[MAX_CMD_LENGTH-1] = '\0';
	    strncpy(arg->pszHost, pDlg->m_pHosts->host, MAX_HOST_LENGTH);
	    arg->pszHost[MAX_HOST_LENGTH-1] = '\0';
	    strcpy(arg->pszJobID, pszJobID);
	    
	    if (pDlg->m_bNoMPI)
	    {
		if (pDlg->m_bUseCommonEnvironment)
		{
		    if (_snprintf(arg->pszEnv, MAX_CMD_LENGTH, "%s", pDlg->m_CommonEnvironment) < 0)
		    {
			// environment variables truncated
			arg->pszEnv[MAX_CMD_LENGTH-1] = '\0';
		    }
		}
		else
		    arg->pszEnv[0] = '\0';
	    }
	    else
	    {
		if (iproc == 0)
		    sprintf(pBuffer, "MPICH_ROOTPORT=-1|MPICH_IPROC=%d|MPICH_SHM_LOW=%d|MPICH_SHM_HIGH=%d", iproc, nShmLow, nShmHigh);
		else
		    sprintf(pBuffer, "MPICH_ROOTPORT=%d|MPICH_IPROC=%d|MPICH_SHM_LOW=%d|MPICH_SHM_HIGH=%d", pDlg->m_nRootPort, iproc, nShmLow, nShmHigh);
		if (strlen(arg->pszEnv) > 0)
		    strncat(arg->pszEnv, "|", MAX_CMD_LENGTH - 1 - strlen(arg->pszEnv));
		if (strlen(pBuffer) + strlen(arg->pszEnv) >= MAX_CMD_LENGTH)
		{
		    // environment variables truncated
		}
		strncat(arg->pszEnv, pBuffer, MAX_CMD_LENGTH - 1 - strlen(arg->pszEnv));
	    }
	    pDlg->m_pProcessThread[iproc] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)MPIRunLaunchProcess, arg, 0, &dwThreadID);
	    if (pDlg->m_pProcessThread[iproc] == NULL)
	    {
		MessageBox(NULL, "Unable to create LaunchProcess thread", "Error", MB_OK);
		// free stuff
		delete arg;
		delete pDlg->m_pProcessThread;
		pDlg->m_nNumProcessThreads = 0;
		pDlg->m_pProcessThread = NULL;
		SetEvent(pDlg->m_hAbortEvent);
		CloseHandle(pDlg->m_hJobThread);
		pDlg->m_hJobThread = NULL;
		pDlg->Abort();
		delete pDlg->m_pProcessSocket;
		delete pDlg->m_pProcessLaunchId;
		delete pDlg->m_pLaunchIdToRank;
		delete pDlg->m_pForwardHost;
		pDlg->m_pProcessSocket = NULL;
		pDlg->m_nNumProcessSockets = 0;
		pDlg->m_pProcessLaunchId = NULL;
		pDlg->m_pLaunchIdToRank = NULL;
		pDlg->m_pForwardHost = NULL;
		pDlg->DisableRunning();
		return;
	    }
	    pDlg->m_nNumProcessThreads++;
	    if (iproc == 0 && !pDlg->m_bNoMPI)
	    {
		// Wait for the root port to be valid
		while (pDlg->m_nRootPort == 0 && (WaitForSingleObject(pDlg->m_hAbortEvent, 0) != WAIT_OBJECT_0))
		    Sleep(200);
		if (pDlg->m_nRootPort == 0)
		{
		    // free stuff
		    CloseHandle(pDlg->m_pProcessThread[0]);
		    delete pDlg->m_pProcessThread;
		    pDlg->m_nNumProcessThreads = 0;
		    pDlg->m_pProcessThread = NULL;
		    CloseHandle(pDlg->m_hJobThread);
		    pDlg->m_hJobThread = NULL;
		    delete pDlg->m_pProcessSocket;
		    delete pDlg->m_pProcessLaunchId;
		    delete pDlg->m_pLaunchIdToRank;
		    delete pDlg->m_pForwardHost;
		    pDlg->m_pProcessSocket = NULL;
		    pDlg->m_nNumProcessSockets = 0;
		    pDlg->m_pProcessLaunchId = NULL;
		    pDlg->m_pLaunchIdToRank = NULL;
		    pDlg->m_pForwardHost = NULL;
		    pDlg->DisableRunning();
		    return;
		}
	    }
	    iproc++;
	}
	
	HostNode *n = pDlg->m_pHosts;
	pDlg->m_pHosts = pDlg->m_pHosts->next;
	delete n;
    }

    // Wait for all the process starting threads to complete
    WaitForLotsOfObjects(pDlg->m_nproc, pDlg->m_pProcessThread);
    pDlg->m_nNumProcessThreads = 0;
    for (i = 0; i<pDlg->m_nproc; i++)
	CloseHandle(pDlg->m_pProcessThread[i]);
    delete pDlg->m_pProcessThread;
    pDlg->m_pProcessThread = NULL;

    if (WaitForSingleObject(pDlg->m_hAbortEvent, 0) == WAIT_OBJECT_0)
    {
	char pszStr[100];
	for (i=0; i<pDlg->m_nproc; i++)
	{
	    if (pDlg->m_pProcessSocket[i]!= INVALID_SOCKET)
	    {
		sprintf(pszStr, "kill %d", pDlg->m_pProcessLaunchId[i]);
		WriteString(pDlg->m_pProcessSocket[i], pszStr);
		//UnmapDrives(pDlg->m_pProcessSocket[i], pDlg->m_pDriveMapList);
		sprintf(pszStr, "freeprocess %d", pDlg->m_pProcessLaunchId[i]);
		WriteString(pDlg->m_pProcessSocket[i], pszStr);
		ReadString(pDlg->m_pProcessSocket[i], pszStr);
		WriteString(pDlg->m_pProcessSocket[i], "done");
		easy_closesocket(pDlg->m_pProcessSocket[i]);
	    }
	}
	delete pDlg->m_pProcessThread;
	pDlg->m_nNumProcessThreads = 0;
	pDlg->m_pProcessThread = NULL;
	CloseHandle(pDlg->m_hJobThread);
	pDlg->m_hJobThread = NULL;
	delete pDlg->m_pProcessSocket;
	delete pDlg->m_pProcessLaunchId;
	delete pDlg->m_pLaunchIdToRank;
	delete pDlg->m_pForwardHost;
	pDlg->m_pProcessSocket = NULL;
	pDlg->m_nNumProcessSockets = 0;
	pDlg->m_pProcessLaunchId = NULL;
	pDlg->m_pLaunchIdToRank = NULL;
	pDlg->m_pForwardHost = NULL;
	pDlg->DisableRunning();
	if (g_bUseJobHost)
	    UpdateJobState("ABORTED");
	return;
    }

    if (g_bUseJobHost)
	UpdateJobState("RUNNING");

    pDlg->WaitForExitCommands();

    delete pDlg->m_pForwardHost;
    pDlg->m_pForwardHost = NULL;

    // Signal the IO redirection thread to stop
    char ch = 0;
    easy_send(pDlg->m_sockStopIOSignalSocket, &ch, 1);

    // Signal all the threads to stop
    SetEvent(pDlg->m_hAbortEvent);

    // Wait for the redirection thread to complete.  Kill it if it takes too long.
    if (WaitForSingleObject(pDlg->m_hRedirectIOListenThread, 10000) != WAIT_OBJECT_0)
    {
	//printf("Terminating the IO redirection control thread\n");
	TerminateThread(pDlg->m_hRedirectIOListenThread, 0);
    }
    CloseHandle(pDlg->m_hRedirectIOListenThread);
    pDlg->m_hRedirectIOListenThread = NULL;
    easy_closesocket(pDlg->m_sockStopIOSignalSocket);

    if (g_bUseJobHost)
	UpdateJobState("FINISHED");

    // Should I free the handle to this thread here?
    CloseHandle(pDlg->m_hJobThread);
    pDlg->m_hJobThread = NULL;

    delete pDlg->m_pProcessSocket;
    delete pDlg->m_pProcessLaunchId;
    delete pDlg->m_pLaunchIdToRank;
    pDlg->m_pProcessSocket = NULL;
    pDlg->m_nNumProcessSockets = 0;
    pDlg->m_pProcessLaunchId = NULL;
    pDlg->m_pLaunchIdToRank = NULL;

    pDlg->DisableRunning();
    }catch(...)
    {
	MessageBox(NULL, "Unhandled exception caught in RunJob thread", "Error", MB_OK);
    }
}

void CachePassword(const char *pszAccount, const char *pszPassword)
{
    int nError;
    char *szEncodedPassword;
    
    TCHAR szKey[256];
    HKEY hRegKey = NULL;
    _tcscpy(szKey, MPICHKEY"\\cache");

    RegDeleteKey(HKEY_CURRENT_USER, szKey);
    if (RegCreateKeyEx(HKEY_CURRENT_USER, szKey,
	0, 
	NULL, 
	REG_OPTION_VOLATILE,
	KEY_ALL_ACCESS, 
	NULL,
	&hRegKey, 
	NULL) != ERROR_SUCCESS) 
    {
	nError = GetLastError();
	//PrintError(nError, "CachePassword:RegDeleteKey(...) failed, error: %d\n", nError);
	return;
    }
    
    // Store the account name
    if (::RegSetValueEx(
	hRegKey, _T("Account"), 0, REG_SZ, 
	(BYTE*)pszAccount, 
	sizeof(TCHAR)*(_tcslen(pszAccount)+1)
	)!=ERROR_SUCCESS)
    {
	nError = GetLastError();
	//PrintError(nError, "CachePassword:RegSetValueEx(...) failed, error: %d\n", nError);
	::RegCloseKey(hRegKey);
	return;
    }

    // encode the password
    szEncodedPassword = EncodePassword((char*)pszPassword);

    // Store the encoded password
    if (::RegSetValueEx(
	hRegKey, _T("Password"), 0, REG_SZ, 
	(BYTE*)szEncodedPassword, 
	sizeof(TCHAR)*(_tcslen(szEncodedPassword)+1)
	)!=ERROR_SUCCESS)
    {
	nError = GetLastError();
	//PrintError(nError, "CachePassword:RegSetValueEx(...) failed, error: %d\n", nError);
	::RegCloseKey(hRegKey);
	free(szEncodedPassword);
	return;
    }

    free(szEncodedPassword);
    ::RegCloseKey(hRegKey);
}

bool ReadCachedPassword(char *pszAccount, char *pszPassword)
{
    int nError;
    char szAccount[100];
    char szPassword[300];
    
    TCHAR szKey[256];
    HKEY hRegKey = NULL;
    _tcscpy(szKey, MPICHKEY"\\cache");

    if (RegOpenKeyEx(HKEY_CURRENT_USER, szKey, 0, KEY_QUERY_VALUE, &hRegKey) == ERROR_SUCCESS) 
    {
	DWORD dwLength = 100;
	*szAccount = TEXT('\0');
	if (RegQueryValueEx(
	    hRegKey, 
	    _T("Account"), NULL, 
	    NULL, 
	    (BYTE*)szAccount, 
	    &dwLength)!=ERROR_SUCCESS)
	{
	    nError = GetLastError();
	    //PrintError(nError, "ReadPasswordFromRegistry:RegQueryValueEx(...) failed, error: %d\n", nError);
	    ::RegCloseKey(hRegKey);
	    return false;
	}
	if (_tcslen(szAccount) < 1)
	    return false;

	*szPassword = '\0';
	dwLength = 300;
	if (RegQueryValueEx(
	    hRegKey, 
	    _T("Password"), NULL, 
	    NULL, 
	    (BYTE*)szPassword, 
	    &dwLength)!=ERROR_SUCCESS)
	{
	    nError = GetLastError();
	    //PrintError(nError, "ReadPasswordFromRegistry:RegQueryValueEx(...) failed, error: %d\n", nError);
	    ::RegCloseKey(hRegKey);
	    return false;
	}

	::RegCloseKey(hRegKey);

	strcpy(pszAccount, szAccount);
	DecodePassword(szPassword);
	strcpy(pszPassword, szPassword);
	return true;
    }

    return false;
}

void CGuiMPIRunView::OnRunBtn()
{
    DWORD dwThreadID;

    UpdateData();

    EnableRunning();

    // Reset the variables generated by a previous run, if there has been one
    m_nRootPort = 0;
    m_bNormalExit = true;
    ResetEvent(m_hAbortEvent);
    ResetEvent(m_hBreakReadyEvent);
    if (!m_bNoClear)
    {
	m_output.SetSel(0, -1);
	m_output.Clear();
    }
    m_bLogon = false;
    m_bFirstBreak = true;
    /*
    while (m_pDriveMapList)
    {
	MapDriveNode *pNode = m_pDriveMapList;
	m_pDriveMapList = m_pDriveMapList->pNext;
	delete pNode;
    }
    */

    if (m_bUseConfigFile)
    {
	if (ParseConfigFile() != PARSE_SUCCESS)
	{
	    MessageBox(m_ConfigFileName, "Error: unable to parse configuration file");
	    DisableRunning();
	    return;
	}
    }
    else
    {
	// Check dialog parameters
	if (m_app.GetLength() < 1)
	{
	    MessageBox("Please specify the application to run", "No executable specified");
	    DisableRunning();
	    return;
	}
	if (!m_bAnyHosts)
	{
	    if (m_host_list.GetSelCount() < 1)
	    {
		MessageBox("Please highlight the hosts you want to launch processes on or choose any hosts.", "No hosts specified");
		DisableRunning();
		return;
	    }
	}
	//CmdLineToUnc(m_app);
    }

    // Get an account/password if necessary
    if (m_bForceLogon)
    {
	CUserPwdDialog dlg;
	dlg.m_remember = FALSE;
	if (dlg.DoModal() != IDOK)
	{
	    MessageBox("No user account supplied", "Aborting application");
	    DisableRunning();
	    return;
	}
	m_Account = dlg.m_account;
	m_Password = dlg.m_password;
	if (dlg.m_remember)
	{
	    if (!SavePasswordToRegistry(m_Account.GetBuffer(0), m_Password.GetBuffer(0), true))
	    {
		DeleteCurrentPasswordRegistryEntry();
	    }
	}
	m_bLogon = true;
    }
    else
    {
	if (m_Account.GetLength() < 1)
	{
	    char pszTemp[10] = "no";
	    ReadMPDRegistry("SingleUser", pszTemp, NULL);
	    if (stricmp(pszTemp, "yes"))
	    {
		if (!ReadCachedPassword(m_Account.GetBuffer(100), m_Password.GetBuffer(100)))
		{
		    HCURSOR hOldCursor = SetCursor( LoadCursor(NULL, IDC_WAIT) );
		    if (!ReadPasswordFromRegistry(m_Account.GetBuffer(100), m_Password.GetBuffer(100)))
		    {
			CUserPwdDialog dlg;
			dlg.m_remember = FALSE;
			if (dlg.DoModal() != IDOK)
			{
			    MessageBox("No user account supplied", "Aborting application");
			    DisableRunning();
			    SetCursor(hOldCursor);
			    return;
			}
			m_Account = dlg.m_account;
			m_Password = dlg.m_password;
			if (dlg.m_remember)
			{
			    if (!SavePasswordToRegistry(m_Account.GetBuffer(0), m_Password.GetBuffer(0), true))
			    {
				DeleteCurrentPasswordRegistryEntry();
			    }
			}
		    }
		    CachePassword(m_Account, m_Password);
		    SetCursor(hOldCursor);
		}
		m_bLogon = true;
	    }
	}
	else
	{
	    m_bLogon = true;
	}
    }

    SaveAppToMRU();

    m_hJobThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)RunJob, this, 0, &dwThreadID);
    if (m_hJobThread == NULL)
    {
	MessageBox("CreateThread(RunJob) failed");
	DisableRunning();
    }
}