mpirun.cpp

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

#include <stdio.h>
#include <tchar.h>
#include "..\Common\MPICH_pwd.h"
#include "..\Common\MPIJobDefs.h"
#include <conio.h>
#include <time.h>
#include "..\Common\Translate_Error.h"
#include "localonly.h"
#include "GetOpt.h"
#include "LaunchProcess.h"
#include "global.h"
#include "WaitThread.h"
#include "MPIRunBNR.h"
#include "Redirection.h"

// Global function pointer declaration
#define BNR_FUNCTION_DECLARATIONS
#include "bnr.h"
#undef BNR_FUNCTION_DECLARATIONS

BNR_Group g_myBNRgroup = BNR_GROUP_NULL;

#define DPRINTF(a) {}
bool LoadBNRFunctions()
{
	HMODULE hBNRLib;
	char pszLibrary[1024];
	
	// First initialize everythting to NULL
	BNR_Init = NULL;
	BNR_Finalize = NULL;
	BNR_Get_group = NULL;
	BNR_Get_parent = NULL;
	BNR_Get_rank = NULL;
	BNR_Get_size = NULL;
	BNR_Open_group = NULL;
	BNR_Close_group = NULL;
	BNR_Free_group = NULL;
	BNR_Merge = NULL;
	BNR_Spawn = NULL;
	BNR_Kill = NULL;
	BNR_Put = NULL;
	BNR_Get = NULL;
	BNR_Fence = NULL;
	BNR_Deposit = NULL;
	BNR_Withdraw = NULL;
	BNR_Lookup = NULL;
	
	BNR_Info_set = NULL;
	BNR_Info_get_valuelen = NULL;
	BNR_Info_get_nthkey = NULL;
	BNR_Info_get_nkeys = NULL;
	BNR_Info_get = NULL;
	BNR_Info_free = NULL;
	BNR_Info_dup = NULL;
	BNR_Info_delete = NULL;
	BNR_Info_create = NULL;

	if (!GetEnvironmentVariable("MPICH_BNR_LIB", pszLibrary, 1024))
	{
		// Try to load the default library:
		strcpy(pszLibrary, "mpichbnr.dll");
		// or just bail out:
		//return false;
	}

	hBNRLib = LoadLibrary(pszLibrary);

	if (hBNRLib == NULL)
		return false;

	// Add code to check if the return values are NULL ...
	BNR_Init = (int (BNR_CALL *)())GetProcAddress(hBNRLib, "BNR_Init");
	if (BNR_Init == NULL) DPRINTF(("BNR_Init == NULL\n"));
	BNR_Finalize = (int (BNR_CALL *)())GetProcAddress(hBNRLib, "BNR_Finallize");
	if (BNR_Finalize == NULL) DPRINTF(("BNR_Finalize == NULL\n"));
	BNR_Get_group = (int (BNR_CALL *)( BNR_Group *mygroup ))GetProcAddress(hBNRLib, "BNR_Get_group");
	if (BNR_Get_group == NULL) DPRINTF(("BNR_Get_group == NULL\n"));
	BNR_Get_parent = (int (BNR_CALL *)( BNR_Group *parent_group ))GetProcAddress(hBNRLib, "BNR_Get_parent");
	if (BNR_Get_parent == NULL) DPRINTF(("BNR_Get_parent == NULL\n"));
	BNR_Get_rank = (int (BNR_CALL *)( BNR_Group group, int *myrank ))GetProcAddress(hBNRLib, "BNR_Get_rank");
	if (BNR_Get_rank == NULL) DPRINTF(("BNR_Get_rank == NULL\n"));
	BNR_Get_size = (int (BNR_CALL *)( BNR_Group group, int *mysize ))GetProcAddress(hBNRLib, "BNR_Get_size");
	if (BNR_Get_size == NULL) DPRINTF(("BNR_Get_size == NULL\n"));
	BNR_Open_group = (int (BNR_CALL *)( BNR_Group local_group, BNR_Group *new_group ))GetProcAddress(hBNRLib, "BNR_Open_group");
	if (BNR_Open_group == NULL) DPRINTF(("BNR_Open_group == NULL\n"));
	BNR_Close_group = (int (BNR_CALL *)( BNR_Group group ))GetProcAddress(hBNRLib, "BNR_Close_group");
	if (BNR_Close_group == NULL) DPRINTF(("BNR_Close_group == NULL\n"));
	BNR_Free_group = (int (BNR_CALL *)( BNR_Group group ))GetProcAddress(hBNRLib, "BNR_Free_group");
	if (BNR_Free_group == NULL) DPRINTF(("BNR_Free_group == NULL\n"));
	BNR_Merge = (int (BNR_CALL *)(BNR_Group local_group, BNR_Group remote_group, BNR_Group *new_group ))GetProcAddress(hBNRLib, "BNR_Merge");
	if (BNR_Merge == NULL) DPRINTF(("BNR_Merge == NULL\n"));
	BNR_Spawn = (int (BNR_CALL *)(BNR_Group remote_group, int count, char *command, char *argv, char *env, BNR_Info info, int (notify_fn)(BNR_Group group, int rank, int exit_code) ))GetProcAddress(hBNRLib, "BNR_Spawn");
	if (BNR_Spawn == NULL) DPRINTF(("BNR_Spawn == NULL\n"));
	BNR_Kill = (int (BNR_CALL *)( BNR_Group group ))GetProcAddress(hBNRLib, "BNR_Kill");
	if (BNR_Kill == NULL) DPRINTF(("BNR_Kill == NULL\n"));
	BNR_Put = (int (BNR_CALL *)( BNR_Group group, char *attr, char *val, int rank_advice ))GetProcAddress(hBNRLib, "BNR_Put");
	if (BNR_Put == NULL) DPRINTF(("BNR_Put == NULL\n"));
	BNR_Get = (int (BNR_CALL *)( BNR_Group group, char *attr, char *val ))GetProcAddress(hBNRLib, "BNR_Get");
	if (BNR_Get == NULL) DPRINTF(("BNR_Get == NULL\n"));
	BNR_Fence = (int (BNR_CALL *)( BNR_Group ))GetProcAddress(hBNRLib, "BNR_Fence");
	if (BNR_Fence == NULL) DPRINTF(("BNR_Fence == NULL\n"));
	BNR_Deposit = (int (BNR_CALL *)( char *attr, char *value ))GetProcAddress(hBNRLib, "BNR_Deposit");
	if (BNR_Deposit == NULL) DPRINTF(("BNR_Deposit == NULL\n"));
	BNR_Withdraw = (int (BNR_CALL *)( char *attr, char *value ))GetProcAddress(hBNRLib, "BNR_Withdraw");
	if (BNR_Withdraw == NULL) DPRINTF(("BNR_Withdraw == NULL\n"));
	BNR_Lookup = (int (BNR_CALL *)( char *attr, char *value ))GetProcAddress(hBNRLib, "BNR_Lookup");
	if (BNR_Lookup == NULL) DPRINTF(("BNR_Lookup == NULL\n"));
	
	BNR_Info_set = (int (BNR_CALL *)(BNR_Info info, char *key, char *value))GetProcAddress(hBNRLib, "BNR_Info_set");
	if (BNR_Info_set == NULL) DPRINTF(("BNR_Info_set == NULL\n"));
	BNR_Info_get_valuelen = (int (BNR_CALL *)(BNR_Info info, char *key, int *valuelen, int *flag))GetProcAddress(hBNRLib, "BNR_Info_get_valuelen");
	if (BNR_Info_get_valuelen == NULL) DPRINTF(("BNR_Info_get_valuelen == NULL\n"));
	BNR_Info_get_nthkey = (int (BNR_CALL *)(BNR_Info info, int n, char *key))GetProcAddress(hBNRLib, "BNR_Info_get_nthkey");
	if (BNR_Info_get_nthkey == NULL) DPRINTF(("BNR_Info_get_nthkey == NULL\n"));
	BNR_Info_get_nkeys = (int (BNR_CALL *)(BNR_Info info, int *nkeys))GetProcAddress(hBNRLib, "BNR_Info_get_nkeys");
	if (BNR_Info_get_nkeys == NULL) DPRINTF(("BNR_Info_get_nkeys == NULL\n"));
	BNR_Info_get = (int (BNR_CALL *)(BNR_Info info, char *key, int valuelen, char *value, int *flag))GetProcAddress(hBNRLib, "BNR_Info_get");
	if (BNR_Info_get == NULL) DPRINTF(("BNR_Info_get == NULL\n"));
	BNR_Info_free = (int (BNR_CALL *)(BNR_Info *info))GetProcAddress(hBNRLib, "BNR_Info_free");
	if (BNR_Info_free == NULL) DPRINTF(("BNR_Info_free == NULL\n"));
	BNR_Info_dup = (int (BNR_CALL *)(BNR_Info info, BNR_Info *newinfo))GetProcAddress(hBNRLib, "BNR_Info_dup");
	if (BNR_Info_dup == NULL) DPRINTF(("BNR_Info_dup == NULL\n"));
	BNR_Info_delete = (int (BNR_CALL *)(BNR_Info info, char *key))GetProcAddress(hBNRLib, "BNR_Info_delete");
	if (BNR_Info_delete == NULL) DPRINTF(("BNR_Info_delete == NULL\n"));
	BNR_Info_create = (int (BNR_CALL *)(BNR_Info *info))GetProcAddress(hBNRLib, "BNR_Info_create");
	if (BNR_Info_create == NULL) DPRINTF(("BNR_Info_create == NULL\n"));

	return true;
}

// Function name	: PrintError
// Description	    : 
// Return type		: void 
// Argument         : HRESULT hr
void PrintError(HRESULT hr)
{
	HLOCAL str;
	FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_ALLOCATE_BUFFER,
		0,
		hr,
		MAKELANGID( LANG_NEUTRAL, SUBLANG_DEFAULT ),
		(LPTSTR) &str,
		0,0);
	
	_tprintf(TEXT("error %d: %s\n"), hr, str);
	LocalFree(str);
}


// Function name	: PrintOptions
// Description	    : 
// Return type		: void 
void PrintOptions()
{
	_tprintf(TEXT("\n"));
	_tprintf(TEXT("Usage:\n"));
	_tprintf(TEXT("   MPIRun configfile [-logon] [args ...]\n"));
	_tprintf(TEXT("   MPIRun -np #processes [-logon] [-env \"var1=val1|var2=val2...\"] executable [args ...]\n"));
	_tprintf(TEXT("   MPIRun -localonly #processes [-env \"var1=val1|var2=val2...\"] exe [args ...]\n"));
	_tprintf(TEXT("\n"));
	_tprintf(TEXT("Config file format:\n"));
	_tprintf(TEXT("   >exe c:\\temp\\mpiprogram.exe\n"));
	_tprintf(TEXT("     OR \\\\host\\share\\mpiprogram.exe\n"));
	_tprintf(TEXT("   >[env var1=val1|var2=val2|var3=val3...]\n"));
	_tprintf(TEXT("   >[args arg1 arg2 ...]\n"));
	_tprintf(TEXT("   >hosts\n"));
	_tprintf(TEXT("   >hostname1 #procs [path\\mpiprogram.exe]\n"));
	_tprintf(TEXT("   >hostname2 #procs [path\\mpiprogram.exe]\n"));
	_tprintf(TEXT("   >hostname3 #procs [path\\mpiprogram.exe]\n"));
	_tprintf(TEXT("   >...\n"));
	_tprintf(TEXT("\n"));
	_tprintf(TEXT("bracketed lines are optional\n"));
	_tprintf(TEXT("\n"));
}

// Function name	: GetHostsFromRegistry
// Description	    : 
// Return type		: bool 
// Argument         : HostNode **list
bool GetHostsFromRegistry(HostNode **list)
{
	DWORD ret_val;
	HKEY hKey;

	// Open the MPICH root key
	if ((ret_val = RegOpenKeyEx(
			HKEY_LOCAL_MACHINE, 
			MPICHKEY,
			0, 
			//KEY_ALL_ACCESS, 
			KEY_QUERY_VALUE,
			&hKey)) != ERROR_SUCCESS)
	{
	    //printf("unable to open mpich registry key\n");fflush(stdout);
		return false;
	}

	// Read the hosts entry
	//TCHAR pszHosts[1024];
	TCHAR *pszHosts = NULL;
	DWORD type, num_bytes=0;//1024*sizeof(TCHAR);
	ret_val = RegQueryValueEx(hKey, _T("Hosts"), 0, &type, NULL, &num_bytes);
	if (ret_val != ERROR_SUCCESS)
	{
	    //printf("unable to query Hosts value length in mpich registry key\n");fflush(stdout);
		return false;
	}
	pszHosts = new TCHAR[num_bytes];
	ret_val = RegQueryValueEx(hKey, _T("Hosts"), 0, &type, (BYTE *)pszHosts, &num_bytes);
	RegCloseKey(hKey);
	if (ret_val != ERROR_SUCCESS)
	{
		delete pszHosts;
		//printf("unable to query Hosts value in mpich registry key\n");fflush(stdout);
		return false;
	}

	TCHAR *token = NULL;
	token = _tcstok(pszHosts, _T("|"));
	if (token != NULL)
	{
		HostNode *n, *l = new HostNode;

		// Make a list of the available nodes
		l->next = NULL;
		_tcscpy(l->host, token);
		l->nSMPProcs = 1;
		n = l;
		while ((token = _tcstok(NULL, _T("|"))) != NULL)
		{
			n->next = new HostNode;
			n = n->next;
			n->next = NULL;
			_tcscpy(n->host, token);
			n->nSMPProcs = 1;
		}
		// add the current host to the end of the list
		n->next = new HostNode;
		n = n->next;
		n->next = NULL;
		_tcscpy(n->host, g_pHosts->host);
		n->nSMPProcs = 1;

		*list = l;

		delete pszHosts;
		return true;
	}

	delete pszHosts;
	return false;
}

// Function name	: GetAvailableHosts
// Description	    : This function requires g_nHosts to have been previously set.
// Return type		: void 
bool GetAvailableHosts()
{
	DWORD size = 100;
	GetComputerName(g_pszFirstHost, &size);
	g_nFirstSMPProcs = 1;
	HostNode *list = NULL;

	// Insert the first host into the list
	g_pHosts = new HostNode;
	_tcscpy(g_pHosts->host, g_pszFirstHost);
	_tcscpy(g_pHosts->exe, g_pszExe);
	g_pHosts->nSMPProcs = 1;
	g_pHosts->next = NULL;

	if (g_nHosts > 1)
	{
		if (GetHostsFromRegistry(&list))
		{
			// add the nodes to the target list, cycling if necessary
			int num_left = g_nHosts-1;
			HostNode *n = list, *target = g_pHosts;
			while (num_left)
			{
				target->next = new HostNode;
				target = target->next;
				target->next = NULL;
				_tcscpy(target->host, n->host);
				_tcscpy(target->exe, g_pHosts->exe);
				target->nSMPProcs = 1;

				n = n->next;
				if (n == NULL)
					n = list;

				num_left--;
			}

			// free the list
			while (list)
			{
				n = list;
				list = list->next;
				delete n;
			}
		}
		else
		{
		    return false;
		    /*
			//printf("Processes will launch locally.\n");
			HostNode *n = g_pHosts;

			for (int i=1; i<g_nHosts; i++)
			{
				n->next = new HostNode;
				_tcscpy(n->next->host, n->host);
				_tcscpy(n->next->exe, n->exe);
				n = n->next;
				n->nSMPProcs = 1;
				n->next = NULL;
			}
			*/
		}
	}
	return true;
}

// Function name	: ParseLineIntoHostNode
// Description	    : 
// Return type		: HostNode* 
// Argument         : LPTSTR line
HostNode* ParseLineIntoHostNode(LPTSTR line)
{
	TCHAR buffer[1024];
	LPTSTR pChar, pChar2;
	HostNode *node = NULL;

	_tcscpy(buffer, line);
	pChar = buffer;

	// Advance over white space
	while (*pChar != _T('\0') && _istspace(*pChar))
		pChar++;
	if (*pChar == _T('#') || *pChar == _T('\0'))
		return NULL;

	// Trim trailing white space
	pChar2 = &buffer[_tcslen(buffer)-1];
	while (_istspace(*pChar2) && (pChar >= pChar))
	{
		*pChar2 = '\0';
		pChar2--;
	}
	
	// If there is anything left on the line, consider it a host name
	if (_tcslen(pChar) > 0)
	{
		node = new HostNode;
		node->nSMPProcs = 1;
		node->next = NULL;
		node->exe[0] = _T('\0');

		// Copy the host name
		pChar2 = node->host;
		while (*pChar != _T('\0') && !_istspace(*pChar))
		{
			*pChar2 = *pChar;
			pChar++;
			pChar2++;
		}
		*pChar2 = _T('\0');