guimpirunview.cpp

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

    CloseHandle(m_hRedirectStdinEvent);
    CloseHandle(m_hRedirectStdinMutex);
    while (m_pRedirectStdinList)
    {
	RedirectStdinStruct *p = m_pRedirectStdinList;
	m_pRedirectStdinList = m_pRedirectStdinList->pNext;
	delete p;
    }

    CloseHandle(m_hBreakReadyEvent);

    easy_socket_finalize();
    CFormView::OnClose();
}

void CGuiMPIRunView::OnAddHostBtn() 
{
    // Add hostname to host list
    UpdateData();
    
    if (m_host.GetLength() != 0)
    {
	
	CString str;
	int n = m_host_list.GetCount();
	if (n != LB_ERR)
	{
	    bool bFound = false;
	    for (int i=0; i<n; i++)
	    {
		m_host_list.GetText(i, str);
		if (str.CompareNoCase(m_host) == 0)
		{
		    bFound = true;
		    break;
		}
	    }
	    if (!bFound)
	    {
		m_host_list.InsertString(-1, m_host);
	    }
	}
    }
}

void CGuiMPIRunView::ReadMRU()
{
    HKEY tkey;
    DWORD result, len;
    DWORD nCount;
    char name[20];
    char value[1024];
    DWORD i;

    m_app_combo.ResetContent();

    // Open the root key
    if (RegCreateKeyEx(HKEY_CURRENT_USER, MPICHKEY,
	0, NULL, REG_OPTION_NON_VOLATILE, KEY_READ, NULL, &tkey, &result) != ERROR_SUCCESS)
    {
	return;
    }

    // Read the count of items
    len = sizeof(DWORD);
    result = RegQueryValueEx(tkey, "mru", 0, NULL, (unsigned char *)&nCount, &len);
    if (result != ERROR_SUCCESS)
    {
	RegCloseKey(tkey);
	return;
    }

    // Read each item into the combo box
    for (i=1; i<=nCount; i++)
    {
	sprintf(name, "mru%d", i);
	len = 1024;
	if (RegQueryValueEx(tkey, name, 0, NULL, (unsigned char *)value, &len) != ERROR_SUCCESS)
	    break;
	m_app_combo.AddString(value);
    }

    RegCloseKey(tkey);
}

void CGuiMPIRunView::SaveAppToMRU()
{
    HKEY tkey;
    DWORD result, len;
    DWORD nCount;
    char value[1024];
    char name[20];
    DWORD i;

    if (m_app_combo.FindString(-1, m_app) != CB_ERR)
	return;
    
    m_app_combo.AddString(m_app);

    // Open the root key
    if (RegCreateKeyEx(HKEY_CURRENT_USER, MPICHKEY,
	0, NULL, REG_OPTION_NON_VOLATILE, KEY_ALL_ACCESS, NULL, &tkey, &result) != ERROR_SUCCESS)
    {
	return;
    }

    // Read the number of mru entries
    len = sizeof(DWORD);
    result = RegQueryValueEx(tkey, "mru", 0, NULL, (unsigned char *)&nCount, &len);
    if (result != ERROR_SUCCESS)
    {
	// If there are none
	nCount = 1;
	RegSetValueEx(tkey, "mru", 0, REG_DWORD, (unsigned char *)&nCount, sizeof(DWORD));
	RegSetValueEx(tkey, "mru1", 0, REG_SZ, (unsigned char *)m_app.GetBuffer(0), m_app.GetLength()+1);
	m_app.ReleaseBuffer();
	RegCloseKey(tkey);
	return;
    }

    if (nCount < (DWORD)m_nMaxMRU)
    {
	nCount++;
	sprintf(name, "mru%d", nCount);
	RegSetValueEx(tkey, name, 0, REG_SZ, (unsigned char *)m_app.GetBuffer(0), m_app.GetLength()+1);
	RegSetValueEx(tkey, "mru", 0, REG_DWORD, (unsigned char *)&nCount, sizeof(DWORD));
	m_app.ReleaseBuffer();
	RegCloseKey(tkey);
	return;
    }

    for (i=1; i<nCount; i++)
    {
	sprintf(name, "mru%d", i+1);
	len = 1024;
	RegQueryValueEx(tkey, name, 0, NULL, (unsigned char *)value, &len);
	sprintf(name, "mru%d", i);
	RegSetValueEx(tkey, name, 0, REG_SZ, (unsigned char *)value, strlen(value)+1);
    }
    sprintf(name, "mru%d", nCount);
    RegSetValueEx(tkey, name, 0, REG_SZ, (unsigned char *)m_app.GetBuffer(0), m_app.GetLength()+1);
    m_app.ReleaseBuffer();

    RegCloseKey(tkey);
}

void CGuiMPIRunView::ClearMRU()
{
    HKEY tkey;
    DWORD result, len;
    DWORD nCount;
    char name[20];
    DWORD i;

    m_app_combo.ResetContent();

    // Open the root key
    if (RegCreateKeyEx(HKEY_CURRENT_USER, MPICHKEY,
	0, NULL, REG_OPTION_NON_VOLATILE, KEY_ALL_ACCESS, NULL, &tkey, &result) != ERROR_SUCCESS)
    {
	return;
    }

    // Read the count of items
    len = sizeof(DWORD);
    result = RegQueryValueEx(tkey, "mru", 0, NULL, (unsigned char *)&nCount, &len);
    if (result != ERROR_SUCCESS)
    {
	nCount = 0;
	RegSetValueEx(tkey, "mru", 0, REG_DWORD, (unsigned char *)&nCount, sizeof(DWORD));
	RegCloseKey(tkey);
	return;
    }

    // Delete each item
    for (i=1; i<=nCount; i++)
    {
	sprintf(name, "mru%d", i);
	RegDeleteValue(tkey, name);
    }

    RegCloseKey(tkey);
}

void CGuiMPIRunView::OnAdvancedBtn() 
{
    CAdvancedOptionsDlg dlg;

    UpdateData();

    dlg.m_bDir = m_bUseWorkingDirectory;
    if (m_bUseWorkingDirectory)
	dlg.m_directory = m_WorkingDirectory;
    dlg.m_bEnv = m_bUseCommonEnvironment;
    dlg.m_environment = m_CommonEnvironment;
    dlg.m_bSlave = m_bUseSlaveProcess;
    dlg.m_slave = m_SlaveProcess;
    dlg.m_bNoClear = m_bNoClear;
    dlg.m_bNoMPI = m_bNoMPI;
    dlg.m_bPassword = m_bForceLogon;
    dlg.m_bConfig = m_bUseConfigFile;
    dlg.m_config_filename = m_ConfigFileName;
    dlg.m_bRedirect = m_redirect;
    dlg.m_output_filename = m_output_filename;
    dlg.m_bNoColor = m_bNoColor;
    dlg.m_bMap = m_bUseMapping;
    dlg.m_map = m_Mappings;
    dlg.m_bCatch = m_bCatch;
    dlg.m_bUseJobHost = g_bUseJobHost;
    if (g_bUseJobHost)
    {
	dlg.m_jobhost = g_pszJobHost;
    }

    if (dlg.DoModal() == IDOK)
    {
	m_bNoColor = dlg.m_bNoColor == TRUE;
	m_bNoClear = dlg.m_bNoClear == TRUE;
	m_bNoMPI = dlg.m_bNoMPI == TRUE;
	m_bForceLogon = dlg.m_bPassword == TRUE;
	m_bCatch = dlg.m_bCatch == TRUE;
	if (dlg.m_bUseJobHost == TRUE && (dlg.m_jobhost.GetLength() > 0))
	{
	    g_bUseJobHost = true;
	    strcpy(g_pszJobHost, dlg.m_jobhost);
	}
	else
	    g_bUseJobHost = false;
	if (dlg.m_bRedirect)
	{
	    m_output_filename = dlg.m_output_filename;
	    m_redirect = true;
	}
	else
	    m_redirect = false;
	if (dlg.m_bConfig)
	{
	    m_bUseCommonEnvironment = false;
	    m_bUseSlaveProcess = false;
	    m_bUseMapping = false;
	    m_bUseWorkingDirectory = false;
	    m_bUseConfigFile = true;
	    m_ConfigFileName = dlg.m_config_filename;
	    m_app_combo.EnableWindow(FALSE);
	    m_app_browse_btn.EnableWindow(FALSE);
	    m_nproc_edit.EnableWindow(FALSE);
	    m_nproc_spin.EnableWindow(FALSE);
	}
	else
	{
	    m_bUseConfigFile = false;
	    m_app_combo.EnableWindow();
	    m_app_browse_btn.EnableWindow();
	    m_nproc_edit.EnableWindow();
	    m_nproc_spin.EnableWindow();
	    if (dlg.m_bEnv)
	    {
		m_CommonEnvironment = dlg.m_environment;
		m_bUseCommonEnvironment = true;
	    }
	    else
		m_bUseCommonEnvironment = false;
	    if (dlg.m_bSlave)
	    {
		m_SlaveProcess = dlg.m_slave;
		m_bUseSlaveProcess = true;
	    }
	    else
		m_bUseSlaveProcess = false;
	    if (dlg.m_bDir)
	    {
		m_WorkingDirectory = dlg.m_directory;
		m_bUseWorkingDirectory = true;
	    }
	    else
		m_bUseWorkingDirectory = false;
	    if (dlg.m_bMap)
	    {
		m_Mappings = dlg.m_map;
		m_bUseMapping = true;
	    }
	    else
		m_bUseMapping = false;
	}
	UpdateData(FALSE);
    }
}

void CGuiMPIRunView::Abort()
{
    SetEvent(m_hAbortEvent);
    if (m_sockBreak != INVALID_SOCKET)
	easy_send(m_sockBreak, "x", 1);
    easy_send(m_sockStopIOSignalSocket, "x", 1);
}

struct ProcessWaitAbortThreadArg
{
    SOCKET sockAbort;
    SOCKET sockStop;
    int n;
    int *pSocket;
};

void ProcessWaitAbort(ProcessWaitAbortThreadArg *pArg)
{
    int n, i;
    fd_set readset;

    FD_ZERO(&readset);
    FD_SET(pArg->sockAbort, &readset);
    FD_SET(pArg->sockStop, &readset);

    n = select(0, &readset, NULL, NULL, NULL);

    if (n == SOCKET_ERROR)
    {
	char str[100];
	sprintf(str, "bselect failed, error %d\n", WSAGetLastError());
	MessageBox(NULL, str, "ProcessWaitAbort", MB_OK);
	for (i=0; i<pArg->n; i++)
	{
	    easy_closesocket(pArg->pSocket[i]);
	}
	easy_closesocket(pArg->sockAbort);
	easy_closesocket(pArg->sockStop);
	return;
    }
    if (n == 0)
    {
	MessageBox(NULL, "bselect returned zero sockets available\n", "ProcessWaitAbort", MB_OK);
	for (i=0; i<pArg->n; i++)
	{
	    easy_closesocket(pArg->pSocket[i]);
	}
	easy_closesocket(pArg->sockAbort);
	easy_closesocket(pArg->sockStop);
	return;
    }
    if (FD_ISSET(pArg->sockAbort, &readset))
    {
	for (i=0; i<pArg->n; i++)
	{
	    easy_send(pArg->pSocket[i], "x", 1);
	}
	if (g_bUseJobHost)
	    UpdateJobState("ABORTED");
    }
    for (i=0; i<pArg->n; i++)
    {
	easy_closesocket(pArg->pSocket[i]);
    }
    easy_closesocket(pArg->sockAbort);
    easy_closesocket(pArg->sockStop);
}

struct ProcessWaitThreadArg
{
    int n;
    SOCKET *pSocket;
    int *pId;
    int *pRank;
    SOCKET sockAbort;
    CGuiMPIRunView *pDlg;
};

void ProcessWait(ProcessWaitThreadArg *pArg)
{
    int i, j, n;
    fd_set totalset, readset;
    char str[256];
    
    FD_ZERO(&totalset);
    
    FD_SET(pArg->sockAbort, &totalset);
    for (i=0; i<pArg->n; i++)
    {
	FD_SET(pArg->pSocket[i], &totalset);
    }
    
    while (pArg->n)
    {
	readset = totalset;
	n = select(0, &readset, NULL, NULL, NULL);
	if (n == SOCKET_ERROR)
	{
	    sprintf(str, "bselect failed, error %d\n", WSAGetLastError());
	    MessageBox(NULL, str, "WaitForExitCommands", MB_OK);
	    for (i=0, j=0; i<pArg->n; i++, j++)
	    {
		while (pArg->pSocket[j] == INVALID_SOCKET)
		    j++;
		easy_closesocket(pArg->pSocket[j]);
	    }
	    return;
	}
	if (n == 0)
	{
	    MessageBox(NULL, "bselect returned zero sockets available", "WaitForExitCommands", MB_OK);
	    for (i=0, j=0; i<pArg->n; i++, j++)
	    {
		while (pArg->pSocket[j] == INVALID_SOCKET)
		    j++;
		easy_closesocket(pArg->pSocket[j]);
	    }
	    return;
	}

	if (FD_ISSET(pArg->sockAbort, &readset))
	{
	    for (i=0; pArg->n > 0; i++)
	    {
		while (pArg->pSocket[i] == INVALID_SOCKET)
		    i++;
		sprintf(str, "kill %d", pArg->pId[i]);
		WriteString(pArg->pSocket[i], str);

		int nRank = pArg->pRank[i];
		if (pArg->pDlg->m_nproc > FORWARD_NPROC_THRESHOLD)
		{
		    if (nRank > 0 && (pArg->pDlg->m_nproc/2) > nRank)
		    {
			//printf("rank %d(%d) stopping forwarder\n", nRank, g_pProcessLaunchId[i]);fflush(stdout);
			sprintf(str, "stopforwarder port=%d abort=yes", pArg->pDlg->m_pForwardHost[nRank].nPort);
			WriteString(pArg->pSocket[i], str);
		    }
		}

		//UnmapDrives(pArg->pSocket[i], pArg->pDlg->m_pDriveMapList);

		sprintf(str, "freeprocess %d", pArg->pId[i]);
		WriteString(pArg->pSocket[i], str);
		ReadString(pArg->pSocket[i], str);