guimpirunview.cpp

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

		WriteString(pArg->pSocket[i], "done");
		easy_closesocket(pArg->pSocket[i]);
		pArg->pSocket[i] = INVALID_SOCKET;
		pArg->n--;
	    }
	    return;
	}
	for (i=0; n>0; i++)
	{
	    while (pArg->pSocket[i] == INVALID_SOCKET)
		i++;
	    if (FD_ISSET(pArg->pSocket[i], &readset))
	    {
		if (!ReadString(pArg->pSocket[i], str))
		{
		    sprintf(str, "Unable to read the result of the getexitcodewait command for process %d, error %d", i, WSAGetLastError());
		    MessageBox(NULL, str, "Critical Error", MB_OK);
		    return;
		}
		
		int nRank = pArg->pRank[i];

		if (strnicmp(str, "FAIL", 4) == 0)
		{
		    sprintf(str, "geterror %d", pArg->pId[i]);
		    WriteString(pArg->pSocket[i], str);
		    ReadString(pArg->pSocket[i], str);
		    printf("getexitcode(rank %d) failed: %s\n", nRank, str);fflush(stdout);
		    if (g_bUseJobHost)
		    {
			UpdateJobKeyValue(nRank, "error", str);
		    
			// get the time the process exited
			sprintf(str, "getexittime %d", pArg->pId[i]);
			WriteString(pArg->pSocket[i], str);
			ReadString(pArg->pSocket[i], str);
			UpdateJobKeyValue(nRank, "exittime", str);
		    }
		    
		    if (easy_send(pArg->sockAbort, "x", 1) == SOCKET_ERROR)
		    {
			printf("Hard abort.\n");fflush(stdout);
			//ExitProcess(-1);
		    }
		}
		else
		{
		    if (g_bUseJobHost)
		    {
			strtok(str, ":"); // strip the extra data from the string
			UpdateJobKeyValue(nRank, "exitcode", str);

			// get the time the process exited
			sprintf(str, "getexittime %d", pArg->pId[i]);
			WriteString(pArg->pSocket[i], str);
			ReadString(pArg->pSocket[i], str);
			UpdateJobKeyValue(nRank, "exittime", str);
		    }
		}

		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?

		sprintf(str, "freeprocess %d", pArg->pId[i]);
		WriteString(pArg->pSocket[i], str);
		ReadString(pArg->pSocket[i], str);
		
		WriteString(pArg->pSocket[i], "done");
		easy_closesocket(pArg->pSocket[i]);
		FD_CLR(pArg->pSocket[i], &totalset);
		pArg->pSocket[i] = INVALID_SOCKET;
		n--;
		pArg->n--;
	    }
	}
    }
}

void CGuiMPIRunView::WaitForExitCommands()
{
    if (m_nNumProcessSockets < FD_SETSIZE)
    {
	int i, j, n;
	fd_set totalset, readset;
	char str[256];
	SOCKET break_sock;
	
	if (m_sockBreak != INVALID_SOCKET)
	    easy_closesocket(m_sockBreak);
	MakeLoop(&break_sock, &m_sockBreak);
	SetEvent(m_hBreakReadyEvent);

	FD_ZERO(&totalset);
	
	FD_SET(break_sock, &totalset);
	for (i=0; i<m_nNumProcessSockets; i++)
	{
	    FD_SET(m_pProcessSocket[i], &totalset);
	}
	
	while (m_nNumProcessSockets)
	{
	    readset = totalset;
	    n = select(0, &readset, NULL, NULL, NULL);
	    if (n == SOCKET_ERROR)
	    {
		sprintf(str, "WaitForExitCommands: bselect failed, error %d\n", WSAGetLastError());
		MessageBox(str);
		for (i=0; m_nNumProcessSockets > 0; i++)
		{
		    while (m_pProcessSocket[i] == INVALID_SOCKET)
			i++;
		    easy_closesocket(m_pProcessSocket[i]);
		    m_nNumProcessSockets--;
		}
		return;
	    }
	    if (n == 0)
	    {
		MessageBox("WaitForExitCommands: bselect returned zero sockets available\n");
		for (i=0; m_nNumProcessSockets > 0; i++)
		{
		    while (m_pProcessSocket[i] == INVALID_SOCKET)
			i++;
		    easy_closesocket(m_pProcessSocket[i]);
		}
		return;
	    }
	    else
	    {
		if (FD_ISSET(break_sock, &readset))
		{
		    int num_read = easy_receive(break_sock, str, 1);
		    if (num_read == 0 || num_read == SOCKET_ERROR)
		    {
			FD_CLR(break_sock, &totalset);
		    }
		    else
		    {
			for (i=0, j=0; i < m_nNumProcessSockets; i++, j++)
			{
			    while (m_pProcessSocket[j] == INVALID_SOCKET)
				j++;
			    sprintf(str, "kill %d", m_pProcessLaunchId[j]);
			    WriteString(m_pProcessSocket[j], str);
			}
		    }
		    n--;
		}
		for (i=0; n>0; i++)
		{
		    while (m_pProcessSocket[i] == INVALID_SOCKET)
			i++;
		    if (FD_ISSET(m_pProcessSocket[i], &readset))
		    {
			if (!ReadString(m_pProcessSocket[i], str))
			{
			    sprintf(str, "Unable to read the result of the getexitcodewait command for process %d, error %d", i, WSAGetLastError());
			    MessageBox(str, "Critical Error", MB_OK);
			    return;
			}
			
			int nRank = m_pLaunchIdToRank[i];

			if (strnicmp(str, "FAIL", 4) == 0)
			{
			    sprintf(str, "geterror %d", m_pProcessLaunchId[i]);
			    WriteString(m_pProcessSocket[i], str);
			    ReadString(m_pProcessSocket[i], str);
			    printf("getexitcode(rank %d) failed: %s\n", nRank, str);fflush(stdout);
			    if (g_bUseJobHost)
			    {
				UpdateJobKeyValue(nRank, "error", str);
			    
				// get the time the process exited
				sprintf(str, "getexittime %d", m_pProcessLaunchId[i]);
				WriteString(m_pProcessSocket[i], str);
				ReadString(m_pProcessSocket[i], str);
				UpdateJobKeyValue(nRank, "exittime", str);
			    }

			    if (easy_send(m_sockBreak, "x", 1) == SOCKET_ERROR)
			    {
				printf("Hard abort.\n");fflush(stdout);
				//ExitProcess(-1);
			    }
			}
			else
			{
			    if (g_bUseJobHost)
			    {
				//printf("ExitProcess: %s\n", str);fflush(stdout);
				strtok(str, ":"); // strip the extra data from the string
				UpdateJobKeyValue(nRank, "exitcode", str);

				// get the time the process exited
				sprintf(str, "getexittime %d", m_pProcessLaunchId[i]);
				WriteString(m_pProcessSocket[i], str);
				ReadString(m_pProcessSocket[i], str);
				UpdateJobKeyValue(nRank, "exittime", str);
			    }
			}

			if (m_nproc > FORWARD_NPROC_THRESHOLD)
			{
			    if (nRank > 0 && (m_nproc/2) > nRank)
			    {
				sprintf(str, "stopforwarder port=%d abort=no", m_pForwardHost[nRank].nPort);
				WriteString(m_pProcessSocket[i], str);
			    }
			}

			//UnmapDrives(m_pProcessSocket[i], m_pDriveMapList);

			sprintf(str, "freeprocess %d", m_pProcessLaunchId[i]);
			WriteString(m_pProcessSocket[i], str);
			ReadString(m_pProcessSocket[i], str);
			
			WriteString(m_pProcessSocket[i], "done");
			easy_closesocket(m_pProcessSocket[i]);
			FD_CLR(m_pProcessSocket[i], &totalset);
			m_pProcessSocket[i] = INVALID_SOCKET;
			n--;
			m_nNumProcessSockets--;
		    }
		}
	    }
	}
	
	easy_closesocket(m_sockBreak);
	m_sockBreak = INVALID_SOCKET;
	delete m_pProcessSocket;
	delete m_pProcessLaunchId;
	delete m_pLaunchIdToRank;
	m_pProcessSocket = NULL;
	m_pProcessLaunchId = NULL;
	m_pLaunchIdToRank = NULL;
    }
    else
    {
	DWORD dwThreadID;
	int num = (m_nNumProcessSockets / (FD_SETSIZE-1)) + 1;
	HANDLE *hThread = new HANDLE[num];
	SOCKET *pAbortsock = new SOCKET[num];
	SOCKET sockStop;
	ProcessWaitThreadArg *arg = new ProcessWaitThreadArg[num];
	ProcessWaitAbortThreadArg *arg2 = new ProcessWaitAbortThreadArg;
	for (int i=0; i<num; i++)
	{
	    if (i == num-1)
		arg[i].n = m_nNumProcessSockets % (FD_SETSIZE-1);
	    else
		arg[i].n = (FD_SETSIZE-1);
	    arg[i].pSocket = &m_pProcessSocket[i*(FD_SETSIZE-1)];
	    arg[i].pId = &m_pProcessLaunchId[i*(FD_SETSIZE-1)];
	    arg[i].pRank = &m_pLaunchIdToRank[i*(FD_SETSIZE-1)];
	    arg[i].pDlg = this;
	    MakeLoop(&arg[i].sockAbort, &pAbortsock[i]);
	}
	for (i=0; i<num; i++)
	{
	    hThread[i] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ProcessWait, &arg[i], 0, &dwThreadID);
	}
	if (m_sockBreak != INVALID_SOCKET)
	    easy_closesocket(m_sockBreak);
	MakeLoop(&arg2->sockAbort, &m_sockBreak);
	MakeLoop(&arg2->sockStop, &sockStop);

	HANDLE hWaitAbortThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ProcessWaitAbort, arg2, 0, &dwThreadID);

	SetEvent(m_hBreakReadyEvent);

	WaitForMultipleObjects(num, hThread, TRUE, INFINITE);
	for (i=0; i<num; i++)
	    CloseHandle(hThread[i]);
	delete hThread;
	delete arg;

	easy_send(sockStop, "x", 1);
	easy_closesocket(sockStop);
	WaitForSingleObject(hWaitAbortThread, 10000);
	delete pAbortsock;
	delete arg2;
	CloseHandle(hWaitAbortThread);

	easy_closesocket(m_sockBreak);
	m_sockBreak = INVALID_SOCKET;
	delete m_pProcessSocket;
	delete m_pProcessLaunchId;
	delete m_pLaunchIdToRank;
	m_pProcessSocket = NULL;
	m_pProcessLaunchId = NULL;
	m_pLaunchIdToRank = NULL;
    }
}

static void ProcessInputString(CString &str)
{
    int index;
    while (true)
    {
	index = str.Find('\b');
	if (index == -1)
	    return;
	str.Delete(index);
	if (index > 0)
	    str.Delete(index-1);
    }
}

void CGuiMPIRunView::OnMsgfilterOutput(NMHDR* pNMHDR, LRESULT* pResult) 
{
    MSGFILTER *pMsgFilter = reinterpret_cast<MSGFILTER *>(pNMHDR);

    if (pMsgFilter->msg == WM_CHAR)
    {
	char ch = (char)pMsgFilter->wParam;
	if (ch == VK_RETURN)
	{
	    ProcessInputString(m_curoutput);
	    m_curoutput += "\r\n";
	    if (WaitForSingleObject(m_hRedirectStdinMutex, 10000) == WAIT_OBJECT_0)
	    {
		if (m_pRedirectStdinList == NULL)
		{
		    m_pRedirectStdinList = new RedirectStdinStruct;
		    m_pRedirectStdinList->str = m_curoutput;
		    m_pRedirectStdinList->pNext = NULL;
		}
		else
		{
		    RedirectStdinStruct *pNode = m_pRedirectStdinList;
		    while (pNode->pNext != NULL)
			pNode = pNode->pNext;
		    pNode->pNext = new RedirectStdinStruct;
		    pNode = pNode->pNext;
		    pNode->pNext = NULL;
		    pNode->str = m_curoutput;
		}
		m_curoutput = "";
		SetEvent(m_hRedirectStdinEvent);
		ReleaseMutex(m_hRedirectStdinMutex);
	    }
	}
	else
	    m_curoutput += ch;
    }

    *pResult = 0;
}

void CGuiMPIRunView::OnResetHostsBtn() 
{
    m_host_list.ResetContent();

    // Get hosts from registry
    char pszHosts[4096];
    DWORD nLength = 4096;
    if (ReadMPDRegistry("hosts", pszHosts, &nLength))
    {
	/*
	char *token = NULL;
	token = strtok(pszHosts, "|");
	if (token != NULL)
	{
	    m_host_list.InsertString(-1, token);
	    while ((token = strtok(NULL, "|")) != NULL)
	    {
		m_host_list.InsertString(-1, token);
	    }
	}
	*/
	QVS_Container *phosts;
	phosts = new QVS_Container(pszHosts);
	if (phosts->first(pszHosts, 4096))
	{
	    m_host_list.InsertString(-1, pszHosts);
	    while (phosts->next(pszHosts, 4096))
	    {
		m_host_list.InsertString(-1, pszHosts);
	    }
	}
	delete phosts;
    }
    else
    {
	char temp[100];
	gethostname(temp, 100);
	m_host_list.InsertString(-1, temp);
    }
}

int CGuiMPIRunView::OnVKeyToItem(UINT nKey, CListBox* pListBox, UINT nIndex) 
{
    if (pListBox == &m_host_list)
    {
	if (nKey == VK_DELETE)
	{
	    int pIndices[1024];
	    int n;
	    n = m_host_list.GetSelItems(1024, pIndices);
	    if (n > 0)
	    {
		for (int i=n-1; i>=0; i--)
		    m_host_list.DeleteString(pIndices[i]);
		if (m_host_list.SetCurSel(pIndices[0]) == LB_ERR)
		    m_host_list.SetCurSel(pIndices[0]-1);
	    }
	    else
	    {
		if (m_host_list.GetCount() > 0)
		{
		    m_host_list.DeleteString(0);
		    m_host_list.SetCurSel(0);
		}
	    }
	}
    }
    
    return CFormView::OnVKeyToItem(nKey, pListBox, nIndex);
}