launchprocess.cpp

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

			hRIThread = CreateThread(
				NULL, 0, 
				(LPTHREAD_START_ROUTINE)RedirectInputThread,
				rarg, 
				0, &dwThreadID);
			
			if (hRIThread == NULL)
			{
				error = GetLastError();
				printf("CreateThread failed: error %d\n", error);
				PrintError(error);
				pLaunch->Release();
				SysFreeString(berror_msg);
				CoUninitialize();
				delete rarg;
				delete arg;
				ExitProcess(1);
			}
		}
		
		IStream *pStream;
		HANDLE hAbortThread = NULL;
		// Marshall pLaunch to a thread which waits for an abort event
		hr = CoMarshalInterThreadInterfaceInStream(IID_IRemoteShell, pLaunch, &pStream);
		if (FAILED(hr))
		{
			pLaunch->Release();
			SysFreeString(berror_msg);
			printf("LaunchProcess:CoMarshalInterThreadInterfaceInStream failed.\n");
			PrintError(hr);
			CoUninitialize();
			delete rarg;
			delete arg;
			ExitProcess(1);
		}
		
		hAbortThread = CreateThread(
			NULL, 0, 
			(LPTHREAD_START_ROUTINE)AbortThread,
			pStream, 
			0, &dwThreadID);
		
		if (hAbortThread == NULL)
		{
			error = GetLastError();
			printf("CreateThread failed: error %d\n", error);
			PrintError(error);
			pLaunch->Release();
			SysFreeString(berror_msg);
			CoUninitialize();
			delete rarg;
			delete arg;
			ExitProcess(1);
		}
		//CloseHandle(hAbortThread);
		if (g_pAbortThreads != NULL)
		    g_pAbortThreads[arg->i] = hAbortThread;
		
		// Redirect output
		HANDLE hStdout = GetStdHandle(STD_OUTPUT_HANDLE);
		HANDLE hStderr = GetStdHandle(STD_ERROR_HANDLE);

		if (hStdout == INVALID_HANDLE_VALUE || hStderr == INVALID_HANDLE_VALUE)
		{
			error = GetLastError();
			_tprintf(TEXT("GetStdHandle failed: Error %d\n"), error);
			pLaunch->Release();
			CoUninitialize();
			SysFreeString(berror_msg);
			if (arg->i == 0) SetEvent(rarg->hEvent);
			delete arg;
			return;
		}

		long more = 1, from = 0, nState = 0;
		VARIANT *v = new VARIANT;
		DWORD num_written, num_elements;
		void *pBuf;
		
		v->vt = VT_UI1 | VT_ARRAY;
		SAFEARRAYBOUND bound;
		bound.lLbound = 0;
		bound.cElements = 0;
		v->parray = SafeArrayCreate(VT_UI1, 1, &bound);
		
		while (more)
		{
			error = 0;
			hr = pLaunch->GetProcessOutput(v, &nState, &error, &berror_msg);
			if (FAILED(hr))
			{
				_tprintf(TEXT("LaunchProcess:GetProcessOutput failed\n"));
				PrintError(hr);
				SysFreeString(berror_msg);
				VariantClear(v);
				delete v;
				if (arg->i == 0) SetEvent(rarg->hEvent);
				delete arg;
				return;
			}
			if (error)
			{
				if (wcslen(berror_msg) < 1)
					wprintf(L"LaunchProcess:GetProcessOutput:Error %d", error);
				else
					wprintf(L"LaunchProcess:%s", berror_msg);
				SysFreeString(berror_msg);
				VariantClear(v);
				delete v;
				if (arg->i == 0) SetEvent(rarg->hEvent);
				delete arg;
				return;
			}
			
			more = nState & RSH_OUTPUT_MORE;

			if (v->parray != NULL)
			{
				num_elements = v->parray->rgsabound->cElements;
				if (num_elements > 0)
				{
					SafeArrayAccessData(v->parray, &pBuf);
					WaitForSingleObject(g_hConsoleOutputMutex, 5000);
#ifdef MULTI_COLOR_OUTPUT
					if (nState & RSH_OUTPUT_STDOUT)
					{
						SetConsoleTextAttribute(hStdout, aConsoleColorAttribute[arg->i%NUM_OUTPUT_COLORS]);
						WriteFile(hStdout, pBuf, num_elements, &num_written, NULL);
						SetConsoleTextAttribute(hStdout, g_ConsoleAttribute);
					}
					else
					{
						SetConsoleTextAttribute(hStderr, aConsoleColorAttribute[arg->i%NUM_OUTPUT_COLORS]);
						WriteFile(hStderr, pBuf, num_elements, &num_written, NULL);
						SetConsoleTextAttribute(hStderr, g_ConsoleAttribute);
					}
#else
					if (nState & RSH_OUTPUT_STDOUT)
						WriteFile(hStdout, pBuf, num_elements, &num_written, NULL);
					else
						WriteFile(hStderr, pBuf, num_elements, &num_written, NULL);
#endif
					FlushFileBuffers(hStdout);
					FlushFileBuffers(hStderr);
					ReleaseMutex(g_hConsoleOutputMutex);
					SafeArrayUnaccessData(v->parray);
					// Destroy the array.
					SafeArrayDestroy(v->parray);
					// Create a new and empty array.
					v->parray = SafeArrayCreate(VT_UI1, 1, &bound);
				}
				else
				{
					SafeArrayDestroy(v->parray);
					v->parray = SafeArrayCreate(VT_UI1, 1, &bound);
				}
			}
			else
			{
				v->parray = SafeArrayCreate(VT_UI1, 1, &bound);
			}
		}
		
		SafeArrayDestroy(v->parray);
		delete v;

		// Stop the redirect input thread
		if (arg->i == 0)
		{
			SetEvent(rarg->hEvent);
			WaitForSingleObject(hRIThread, 5000);
			CloseHandle(hRIThread);
		}

		pLaunch->Release();

		CoUninitialize();
	}
	else
	{
		wprintf(L"Connect to %s failed\n", arg->pszHost);
		if (arg->i == 0)
			ExitProcess(1);
	}
		
	delete arg;
}

// Function name	: LaunchProcessWithMSH
// Description	    : 
// Return type		: void 
// Argument         : LaunchProcessArg *arg
void LaunchProcessWithMSH(LaunchProcessArg *arg)
{
	IRemoteShell *pLaunch;
	long error;
	BSTR berror_msg;
	BSTR bFilename;
	HRESULT hr;
	
	if (arg->i == 0)
	{
		hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
		if (FAILED(hr))
		{
			printf("CoInitializeEx failed: ");
			PrintError(hr);
			delete arg;
			return;
		}
		
		// Connect to RemoteShell
		if (!Connect(arg->pszHost, pLaunch))
		{
			wprintf(L"Connect to %s failed\n", arg->pszHost);
			ExitProcess(1);
		}
		berror_msg = SysAllocString(L"");
		bFilename = SysAllocString(L"");
		error = 0;
		hr = pLaunch->CreateTempFile(&bFilename, &error, &berror_msg);
		if (FAILED(hr))
		{
			wprintf(L"LaunchProcess:CreateTempFile failed on %s\n", arg->pszHost);
			PrintError(hr);
			pLaunch->Release();
			delete arg;
			SysFreeString(bFilename);
			ExitProcess(1);
		}
		if (error)
		{
			pLaunch->Release();
			wprintf(L"LaunchProces:CreateTempFile failed on %s: %s", arg->pszHost, berror_msg);
			delete arg;
			SysFreeString(bFilename);
			ExitProcess(1);
		}
		wcscat(arg->pszEnv, L"|MPICH_EXTRA=");
		wcscat(arg->pszEnv, bFilename);
	}
	
	WCHAR wCmdLine[1024];
	swprintf(wCmdLine, L"msh ");
	if (arg->bLogon)
	{
		wcscat(wCmdLine, L"-account \"");
		wcscat(wCmdLine, arg->pszAccount);
		wcscat(wCmdLine, L"\" -password \"");
		wcscat(wCmdLine, arg->pszPassword);
		wcscat(wCmdLine, L"\" ");
	}
	if (wcslen(arg->pszEnv))
	{
		wcscat(wCmdLine, L"-env \"");
		wcscat(wCmdLine, arg->pszEnv);
		wcscat(wCmdLine, L"\" ");
	}
	if (wcslen(arg->pszDir))
	{
		wcscat(wCmdLine, L"-dir \"");
		wcscat(wCmdLine, arg->pszDir);
		wcscat(wCmdLine, L"\" ");
	}
	wcscat(wCmdLine, arg->pszHost);
	wcscat(wCmdLine, L" ");
	wcscat(wCmdLine, arg->pszCmdLine);
	
	PROCESS_INFORMATION psInfo;
	STARTUPINFOW saInfo;
	memset(&saInfo, 0, sizeof(STARTUPINFO));
	saInfo.cb         = sizeof(STARTUPINFO);
	if (CreateProcessW(
		NULL,
		wCmdLine,
		NULL, NULL, TRUE,
		0,
		//DETACHED_PROCESS | IDLE_PRIORITY_CLASS, 
		//CREATE_NO_WINDOW | IDLE_PRIORITY_CLASS,
		//CREATE_NO_WINDOW | IDLE_PRIORITY_CLASS | CREATE_NEW_PROCESS_GROUP,
		//DETACHED_PROCESS | IDLE_PRIORITY_CLASS | CREATE_NEW_PROCESS_GROUP,
		//CREATE_NO_WINDOW | IDLE_PRIORITY_CLASS | CREATE_SUSPENDED, 
		NULL,
		NULL,
		&saInfo, &psInfo))
	{
		//wprintf(L"'%s'\n", wCmdLine);
		CloseHandle(psInfo.hThread);
	}
	else
		ExitProcess(1);
	
	if (arg->i == 0)
	{
		// Get the port number and redirect input to the first process
		error = 0;
		hr = pLaunch->GetPortFromFile(bFilename, &g_nRootPort, &error, &berror_msg);
		SysFreeString(bFilename);
		if (FAILED(hr))
		{
			pLaunch->Release();
			printf("LaunchProcess:GetPortFromFile failed\n");
			delete arg;
			ExitProcess(1);
		}
		if (error)
		{
			pLaunch->Release();
			wprintf(L"LaunchProcess:%s", berror_msg);
			delete arg;
			ExitProcess(1);
		}
		
		pLaunch->Release();
		
		CoUninitialize();
	}
	
	WaitForSingleObject(psInfo.hProcess, INFINITE);
	CloseHandle(psInfo.hProcess);
	
	delete arg;
}