mpirun.cpp

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

#include <stdio.h>
#include "..\Common\MPICH_pwd.h"
#include "..\Common\MPIJobDefs.h"
#include <conio.h>
#include <time.h>
#include "Translate_Error.h"
#include "localonly.h"
#include "GetOpt.h"
#include "LaunchProcess.h"
#include "global.h"
#include "WaitThread.h"
#include "mpd.h"
#include "mpdutil.h"
#include "RedirectIO.h"
#include <ctype.h>
#include <stdlib.h>
#include "mpirun.h"
#include "parsecliques.h"

// Prototypes
void ExeToUnc(char *pszExe);

void PrintError(int error, char *msg, ...)
{
    int n;
    va_list list;
    HLOCAL str;
    int num_bytes;

    va_start(list, msg);
    n = vprintf(msg, list);
    va_end(list);
    
    num_bytes = FormatMessage(
	FORMAT_MESSAGE_FROM_SYSTEM |
	FORMAT_MESSAGE_ALLOCATE_BUFFER,
	0,
	error,
	MAKELANGID( LANG_NEUTRAL, SUBLANG_DEFAULT ),
	(LPTSTR) &str,
	0,0);
    
    printf("Error: %s", str);
    fflush(stdout);
    
    LocalFree(str);
}

// Function name	: PrintOptions
// Description	    : 
// Return type		: void 
void PrintOptions()
{
    printf("\n");
    printf("Usage:\n");
    printf("   MPIRun -np #processes [options] executable [args ...]\n");
    printf("   MPIRun [options] configfile [args ...]\n");
    printf("\n");
    printf("mpirun options:\n");
    printf("   -localonly\n");
    printf("   -env \"var1=val1|var2=val2|var3=val3...\"\n");
    printf("   -dir drive:\\my\\working\\directory\n");
    printf("   -map drive:\\\\host\\share\n");
    printf("   -logon\n");
    printf("\n");
    printf("Config file format:\n");
    printf("   >exe c:\\temp\\mpiprogram.exe\n");
    printf("     OR \\\\host\\share\\mpiprogram.exe\n");
    printf("   >[env var1=val1|var2=val2|var3=val3...]\n");
    printf("   >[dir drive:\\my\\working\\directory]\n");
    printf("   >[map drive:\\\\host\\share]\n");
    printf("   >[args arg1 arg2 ...]\n");
    printf("   >hosts\n");
    printf("   >hostname1 #procs [path\\mpiprogram.exe]\n");
    printf("   >hostname2 #procs [path\\mpiprogram.exe]\n");
    printf("   >hostname3 #procs [path\\mpiprogram.exe]\n");
    printf("   >...\n");
    printf("\n");
    printf("bracketed lines are optional\n");
    printf("\n");
    printf("For a list of all mpirun options, execute 'mpirun -help2'\n");
    printf("\n");
}

void PrintExtraOptions()
{
    printf("\n");
    printf("All options to mpirun:\n");
    printf("\n");
    printf("-np x\n");
    printf("  launch x processes\n");
    printf("-localonly x\n");
    printf("-np x -localonly\n");
    printf("  launch x processes on the local machine\n");
    printf("-machinefile filename\n");
    printf("  use a file to list the names of machines to launch on\n");
    printf("-hosts n host1 host2 ... hostn\n");
    printf("-hosts n host1 m1 host2 m2 ... hostn mn\n");
    printf("  launch on the specified hosts\n");
    printf("  the number of processes = m1 + m2 + ... + mn\n");
    printf("-map drive:\\\\host\\share\n");
    printf("  map a drive on all the nodes\n");
    printf("  this mapping will be removed when the processes exit\n");
    printf("-dir drive:\\my\\working\\directory\n");
    printf("  launch processes in the specified directory\n");
    printf("-env \"var1=val1|var2=val2|var3=val3...\"\n");
    printf("  set environment variables before launching the processes\n");
    printf("-logon\n");
    printf("  prompt for user account and password\n");
    printf("-pwdfile filename\n");
    printf("  read the account and password from the file specified\n");
    printf("  put the account on the first line and the password on the second\n");
    printf("-tcp\n");
    printf("  use tcp instead of shared memory on the local machine\n");
    printf("-getphrase\n");
    printf("  prompt for the passphrase to access remote mpds\n");
    printf("-nocolor\n");
    printf("  don't use process specific output coloring\n");
    printf("-nompi\n");
    printf("  launch processes without the mpi startup mechanism\n");
    printf("-nodots\n");
    printf("  don't output dots while logging on the user\n");
    printf("-nomapping\n");
    printf("  don't try to map the current directory on the remote nodes\n");
    printf("-nopopup_debug\n");
    printf("  disable the system popup dialog if the process crashes\n");
    printf("-dbg\n");
    printf("  catch unhandled exceptions\n");
    printf("-jobhost hostname\n");
    printf("  send job information to the specified host\n");
    printf("-jobhostmpdpwd passphrase\n");
    printf("  specify the jobhost passphrase\n");
    printf("-exitcodes\n");
    printf("  print the process exit codes when each process exits.\n");
    printf("-noprompt\n");
    printf("  prevent mpirun from prompting for user credentials.\n");
    printf("-priority class[:level]\n");
    printf("  set the process startup priority class and optionally level.\n");
    printf("  class = 0,1,2,3,4   = idle, below, normal, above, high\n");
    printf("  level = 0,1,2,3,4,5 = idle, lowest, below, normal, above, highest\n");
    printf("  the default is -priority 1:3\n");
    printf("-mpduser\n");
    printf("  use the installed mpd single user ignoring the current user credentials.\n");
}

// Function name	: ConnectReadMPDRegistry
// Description	    : 
// Return type		: bool 
// Argument         : char *pszHost
// Argument         : int nPort
// Argument         : char *pszPassPhrase
// Argument         : char *name
// Argument         : char *value
// Argument         : DWORD *length = NULL
bool ConnectReadMPDRegistry(char *pszHost, int nPort, char *pszPassPhrase, char *name, char *value, DWORD *length = NULL)
{
    int error;
    SOCKET sock;
    char pszStr[1024];

    if ((error = ConnectToMPD(pszHost, nPort, pszPassPhrase, &sock)) == 0)
    {
	sprintf(pszStr, "lget %s", name);
	WriteString(sock, pszStr);
	ReadStringTimeout(sock, pszStr, g_nMPIRUN_SHORT_TIMEOUT);
	if (strlen(pszStr))
	{
	    WriteString(sock, "done");
	    easy_closesocket(sock);
	    strcpy(value, pszStr);
	    if (length)
		*length = strlen(pszStr);
	    //printf("ConnectReadMPDRegistry successfully used to read the host entry:\n%s\n", pszStr);fflush(stdout);
	    return true;
	}
	WriteString(sock, "done");
	easy_closesocket(sock);
    }
    else
    {
	//printf("MPIRunLaunchProcess: Connect to %s failed, error %d\n", pszHost, error);fflush(stdout);
    }
    return false;
}

// Function name	: ReadMPDRegistry
// Description	    : 
// Return type		: bool 
// Argument         : char *name
// Argument         : char *value
// Argument         : DWORD *length = NULL
bool ReadMPDRegistry(char *name, char *value, DWORD *length /*= NULL*/)
{
    HKEY tkey;
    DWORD len, result;
    
    // Open the root key
    if (RegOpenKeyEx(HKEY_LOCAL_MACHINE, MPD_REGISTRY_KEY,
	0, 
	KEY_READ,
	&tkey) != ERROR_SUCCESS)
    {
	//printf("Unable to open the MPD registry key, error %d\n", GetLastError());
	return false;
    }
    
    if (length == NULL)
	len = MAX_CMD_LENGTH;
    else
	len = *length;
    result = RegQueryValueEx(tkey, name, 0, NULL, (unsigned char *)value, &len);
    if (result != ERROR_SUCCESS)
    {
	//printf("Unable to read the mpd registry key '%s', error %d\n", name, GetLastError());
	RegCloseKey(tkey);
	return false;
    }
    if (length != NULL)
	*length = len;
    
    RegCloseKey(tkey);
    return true;
}

// Function name	: ReadCachedPassword
// Description	    : 
// Return type		: bool 
bool ReadCachedPassword()
{
    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 ((nError = RegQueryValueEx(
	    hRegKey, 
	    _T("Account"), NULL, 
	    NULL, 
	    (BYTE*)szAccount, 
	    &dwLength))!=ERROR_SUCCESS)
	{
	    //PrintError(nError, "ReadPasswordFromRegistry:RegQueryValueEx(...) failed, error: %d\n", nError);
	    ::RegCloseKey(hRegKey);
	    return false;
	}
	if (_tcslen(szAccount) < 1)
	    return false;

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

	::RegCloseKey(hRegKey);

	strcpy(g_pszAccount, szAccount);
	DecodePassword(szPassword);
	strcpy(g_pszPassword, szPassword);
	return true;
    }

    return false;
}

// Function name	: CachePassword
// Description	    : 
// Return type		: void
void CachePassword()
{
    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 ((nError = ::RegSetValueEx(
	hRegKey, _T("Account"), 0, REG_SZ, 
	(BYTE*)g_pszAccount, 
	sizeof(TCHAR)*(_tcslen(g_pszAccount)+1)
	))!=ERROR_SUCCESS)
    {
	//PrintError(nError, "CachePassword:RegSetValueEx(%s) failed, error: %d\n", g_pszAccount, nError);
	::RegCloseKey(hRegKey);
	return;
    }

    // encode the password
    szEncodedPassword = EncodePassword(g_pszPassword);

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

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

// Function name	: GetHostsFromRegistry
// Description	    : 
// Return type		: bool 
// Argument         : HostNode **list
bool GetHostsFromRegistry(HostNode **list)
{
    // Read the hosts entry
    char pszHosts[MAX_CMD_LENGTH+1];
    DWORD nLength = MAX_CMD_LENGTH;
    if (!ReadMPDRegistry("hosts", pszHosts, &nLength))
    {
	char localhost[100];
	gethostname(localhost, 100);
	if (!ConnectReadMPDRegistry(localhost, MPD_DEFAULT_PORT, MPD_DEFAULT_PASSPHRASE, "hosts", pszHosts, &nLength))
	    return false;
    }

    QVS_Container *phosts;
    phosts = new QVS_Container(pszHosts);
    if (phosts->first(pszHosts, MAX_CMD_LENGTH))
    {
	HostNode *n, *l = new HostNode;
	
	l->next = NULL;
	strncpy(l->host, pszHosts, MAX_HOST_LENGTH);
	l->host[MAX_HOST_LENGTH-1] = '\0';
	l->nSMPProcs = 1;
	n = l;
	while (phosts->next(pszHosts, MAX_CMD_LENGTH))
	{
	    n->next = new HostNode;
	    n = n->next;
	    n->next = NULL;
	    strncpy(n->host, pszHosts, MAX_HOST_LENGTH);
	    n->host[MAX_HOST_LENGTH-1] = '\0';
	    n->nSMPProcs = 1;
	}
	*list = l;
	delete phosts;
	return true;
    }
    delete phosts;
    return false;
}

// Function name	: GetAvailableHosts
// Description	    : This function requires g_nHosts to have been previously set.
// Return type		: bool 
bool GetAvailableHosts()
{
    HostNode *list = NULL;
    
    //dbg_printf("finding available hosts\n");
    if (g_nHosts > 0)
    {
	if (GetHostsFromRegistry(&list))
	{
	    // Insert the first host into the list
	    g_pHosts = new HostNode;
	    strncpy(g_pHosts->host, list->host, MAX_HOST_LENGTH);
	    g_pHosts->host[MAX_HOST_LENGTH-1] = '\0';
	    strncpy(g_pHosts->exe, g_pszExe, MAX_CMD_LENGTH);
	    g_pHosts->exe[MAX_CMD_LENGTH-1] = '\0';
	    g_pHosts->nSMPProcs = 1;
	    g_pHosts->next = NULL;

	    // add the nodes to the target list, cycling if necessary
	    int num_left = g_nHosts-1;
	    HostNode *n = list->next, *target = g_pHosts;
	    if (n == NULL)
		n = list;
	    while (num_left)
	    {
		target->next = new HostNode;
		target = target->next;
		target->next = NULL;
		strncpy(target->host, n->host, MAX_HOST_LENGTH);
		target->host[MAX_HOST_LENGTH-1] = '\0';
		strncpy(target->exe, g_pHosts->exe, MAX_CMD_LENGTH);
		target->exe[MAX_CMD_LENGTH-1] = '\0';
		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;
	}
    }
    return true;
}

// Function name	: GetHostsFromFile
// Description	    : 
// Return type		: bool 
// Argument         : char *pszFileName
bool GetHostsFromFile(char *pszFileName)
{
    FILE *fin;
    char buffer[1024] = "";
    char *pChar, *pChar2;
    HostNode *node = NULL, *list = NULL, *cur_node;

    fin = fopen(pszFileName, "r");
    if (fin == NULL)
    {
	printf("unable to open file '%s'\n", pszFileName);
	return false;
    }
    
    // Read the host names from the file
    while (fgets(buffer, 1024, fin))
    {
	pChar = buffer;
	
	// Advance over white space
	while (*pChar != '\0' && isspace(*pChar))
	    pChar++;
	if (*pChar == '#' || *pChar == '\0')
	    continue;
	
	// Trim trailing white space
	pChar2 = &buffer[strlen(buffer)-1];
	while (isspace(*pChar2) && (pChar >= pChar))
	{
	    *pChar2 = '\0';
	    pChar2--;
	}
	
	// If there is anything left on the line, consider it a host name
	if (strlen(pChar) > 0)
	{
	    node = new HostNode;
	    node->nSMPProcs = 1;
	    node->next = NULL;
	    node->exe[0] = '\0';
	    
	    // Copy the host name
	    pChar2 = node->host;
	    while (*pChar != '\0' && !isspace(*pChar))
	    {
		*pChar2 = *pChar;
		pChar++;
		pChar2++;
	    }
	    *pChar2 = '\0';
	    pChar2 = strtok(node->host, ":");
	    pChar2 = strtok(NULL, "\n");
	    if (pChar2 != NULL)
	    {
		node->nSMPProcs = atoi(pChar2);
		if (node->nSMPProcs < 1)
		    node->nSMPProcs = 1;
	    }
	    else
	    {
		// Advance over white space
		while (*pChar != '\0' && isspace(*pChar))
		    pChar++;
		// Get the number of SMP processes
		if (*pChar != '\0')
		{
		    node->nSMPProcs = atoi(pChar);
		    if (node->nSMPProcs < 1)
			node->nSMPProcs = 1;
		}
	    }

	    if (list == NULL)
	    {
		list = node;
		cur_node = node;
	    }
	    else
	    {
		cur_node->next = node;
		cur_node = node;
	    }
	}
    }

    fclose(fin);

    if (list == NULL)
	return false;

    // Allocate the first host node
    g_pHosts = new HostNode;
    int num_left = g_nHosts;
    HostNode *n = list, *target = g_pHosts;

    // add the nodes to the target list, cycling if necessary
    while (num_left)