mpdleft.cpp

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

    {
	GetStringOpt(p->pszIn, "dest", pszHost);
	if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	{
	    char pszId[10];
	    char pszPid[10];
	    GetStringOpt(p->pszIn, "id", pszId);
	    if (GetStringOpt(p->pszIn, "pid", pszPid))
	    {
		SavePid(atoi(pszId), atoi(pszPid));
	    }
	    else
	    {
		if (GetStringOpt(p->pszIn, "error", pszStr))
		{
		    SaveError(atoi(pszId), pszStr);
		}
	    }
	}
	else
	{
	    GetStringOpt(p->pszIn, "src", pszHost);
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		err_printf("launched result went full circle, discarding\n");
	    }
	    else
	    {
		// forward the message along
		ContextWriteString(g_pRightContext, p->pszIn);
	    }
	}
    }
    else if (strnicmp(p->pszIn, "exitcode ", 9) == 0)
    {
	GetStringOpt(p->pszIn, "dest", pszHost);
	if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	{
	    int id = 0;
	    char pszId[10];
	    char pszExitCode[20];
	    char pszError[256];
	    char timestamp[256];
	    if (GetStringOpt(p->pszIn, "id", pszId))
		id = atoi(pszId);
	    GetStringOpt(p->pszIn, "code", pszExitCode);
	    if (GetStringOpt(p->pszIn, "time", timestamp))
		SaveTimestamp(id, timestamp);
	    else
		SaveTimestamp(id, "unknown");
	    if (GetStringOpt(p->pszIn, "error", pszError))
	    {
		SaveError(id, pszError);
	    }
	    else
	    {
		SaveExitCode(id, atoi(pszExitCode));
	    }
	}
	else
	{
	    GetStringOpt(p->pszIn, "src", pszHost);
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		err_printf("exitcode result went full circle, discarding\n");
	    }
	    else
	    {
		// forward the message along
		ContextWriteString(g_pRightContext, p->pszIn);
	    }
	}
    }
    else if (strnicmp(p->pszIn, "hosts ", 6) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost))
	{
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		char pszBFD[10];
		MPD_Context *pContext;
		
		GetStringOpt(p->pszIn, "result", pszStr);
		GetStringOpt(p->pszIn, "sock", pszBFD);
		pContext = GetContext(atoi(pszBFD));
		if (pContext != NULL)
		{
		    ContextWriteString(pContext, pszStr);
		}
		else
		{
		    err_printf("console context not found\n");
		}
	    }
	    else
	    {
		strncpy(pszStr, p->pszIn, MAX_CMD_LENGTH);
		pszStr[MAX_CMD_LENGTH-1] = '\0';
		strncat(pszStr, ",", MAX_CMD_LENGTH - 1 - strlen(pszStr));
		strncat(pszStr, g_pszHost, MAX_CMD_LENGTH - 1 - strlen(pszStr));
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    err_printf("invalid hosts command '%s' read\n", p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "next ", 5) == 0)
    {
	char pszBfd[10];
	char pszN[10] = "0";
	int n;
	GetStringOpt(p->pszIn, "n", pszN);
	GetStringOpt(p->pszIn, "src", pszHost);
	GetStringOpt(p->pszIn, "sock", pszBfd);
	n = atoi(pszN);
	if ((n > 0) && (n < 16384))
	{
	    n--;
	    _snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=%s", g_pszHost, pszHost, pszBfd, g_pszHost);
	    ContextWriteString(g_pRightContext, pszStr);
	    if (n > 0)
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "next src=%s sock=%s n=%d", pszHost, pszBfd, n);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    _snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=Error: invalid number of hosts requested", g_pszHost, pszHost, pszBfd);
	    ContextWriteString(g_pRightContext, pszStr);
	}
    }
    else if (strnicmp(p->pszIn, "barrier ", 8) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost))
	{
	    if (strcmp(pszHost, g_pszHost) && strcmp(pszHost, g_pszIP))
	    {
		char pszName[100], pszCount[10];
		if (GetStringOpt(p->pszIn, "name", pszName))
		{
		    if (GetStringOpt(p->pszIn, "count", pszCount))
		    {
			SetBarrier(pszName, atoi(pszCount), INVALID_SOCKET);
			ContextWriteString(g_pRightContext, p->pszIn);
		    }
		}
	    }
	}
    }
    else if (strnicmp(p->pszIn, "ps ", 3) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost))
	{
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		char pszBFD[10];
		MPD_Context *pContext;
		
		if (GetStringOpt(p->pszIn, "result", pszStr))
		{
		    // Chop off the trailing carriage return
		    int nLength = strlen(pszStr);
		    if (nLength > 2)
		    {
			if (pszStr[nLength-1] == '\r' || pszStr[nLength-1] == '\n')
			    pszStr[nLength-1] = '\0';
			if (pszStr[nLength-2] == '\r' || pszStr[nLength-2] == '\n')
			    pszStr[nLength-2] = '\0';
		    }
		}
		GetStringOpt(p->pszIn, "sock", pszBFD);
		pContext = GetContext(atoi(pszBFD));
		if (pContext != NULL)
		{
		    ContextWriteString(pContext, pszStr);
		}
		else
		{
		    err_printf("console context not found\n");
		}
	    }
	    else
	    {
		strncpy(pszStr, p->pszIn, MAX_CMD_LENGTH);
		pszStr[MAX_CMD_LENGTH-1] = '\0';
		ConcatenateProcessesToString(pszStr);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    err_printf("invalid ps command '%s' read\n", p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "lefthost ", 9) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost) && GetStringOpt(p->pszIn, "host", g_pszInsertHost))
	{
	    WriteMPDRegistry(INSERT1, g_pszInsertHost);
	    
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "leftlefthost src=%s host=%s", g_pszHost, g_pszInsertHost);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	    else
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "lefthost src=%s host=%s", pszHost, g_pszHost);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    err_printf("invalid lefthost command '%s'\n", p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "leftlefthost ", 13) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost) && GetStringOpt(p->pszIn, "host", g_pszInsertHost2))
	{
	    WriteMPDRegistry(INSERT2, g_pszInsertHost2);
	    
	    if ((strcmp(pszHost, g_pszHost)) && (strcmp(pszHost, g_pszIP)))
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "leftlefthost src=%s host=%s", pszHost, g_pszInsertHost);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	}
	else
	{
	    err_printf("invalid lefthost command '%s'\n", p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "kill ", 5) == 0)
    {
	if (GetStringOpt(p->pszIn, "host", pszHost))
	{
	    if ((strcmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		char pszPid[10];
		if (GetStringOpt(p->pszIn, "pid", pszPid))
		{
		    int nPid = atoi(pszPid);
		    //dbg_printf("MPD_KillProcess host=%s pid=%d\n", g_pszHost, nPid);
		    MPD_KillProcess(nPid);
		}
	    }
	    else
	    {
		if (GetStringOpt(p->pszIn, "src", pszStr))
		{
		    if (stricmp(pszStr, g_pszHost) == 0)
		    {
			char pszTry[10], pszPid[10];
			if (!GetStringOpt(p->pszIn, "try", pszTry))
			{
			    if (GetStringOpt(p->pszIn, "pid", pszPid) && GetIPString(pszStr, pszHost))
			    {
				// try the kill command again with the ip string instead of the host name
				_snprintf(pszStr, MAX_CMD_LENGTH, "kill src=%s host=%s pid=%s try=2", g_pszHost, pszHost, pszPid);
				ContextWriteString(g_pRightContext, pszStr);
			    }
			}
			else
			{
			    dbg_printf("kill command went full circle without matching any hosts, '%s'\n", p->pszIn);
			}
		    }
		}
		else
		{
		    err_printf("kill command has no source, '%s'\n", p->pszIn);
		}
	    }
	}
    }
    else if (strnicmp(p->pszIn, "killall ", 8) == 0)
    {
	if (GetStringOpt(p->pszIn, "src", pszHost))
	{
	    ShutdownAllProcesses();
	    // Conceivably there could be forwarders at this node not associated with processes in this ring.
	    // In that case you should not call AbortAllForwarders(), but I am not going to figure out how to track that
	    // case any time soon.
	    AbortAllForwarders();
	    if ((strcmp(pszHost, g_pszHost) != 0) && (strcmp(pszHost, g_pszIP) != 0))
	    {
		ContextWriteString(g_pRightContext, p->pszIn);
	    }
	}
	else
	{
	    err_printf("invalid killall command '%s' read\n", p->pszIn);
	}
    }
    else if (stricmp(p->pszIn, "exitall") == 0)
    {
	if (g_bExitAllRoot)
	{
	    RemoveContext(g_pRightContext);
	    g_pRightContext = NULL;
	}
	else
	{
	    ContextWriteString(g_pRightContext, "exitall");
	}
	p->nState = MPD_INVALID;
	p->bDeleteMe = true;
	SignalExit();
	SignalExit(); // Signal twice to get the service to stop
    }
    else if (stricmp(p->pszIn, "done") == 0)
    {
	dbg_printf("left[%d] read 'done'\n", p->sock);
	p->nState = MPD_INVALID;
	p->bDeleteMe = true;
    }
    else if (stricmp(p->pszIn, "new left") == 0)
    {
	if (p == g_pLeftContext)
	{
	    err_printf("Error, current left thread context read 'new left' command\n");
	}
	// save the old left host
	strcpy(pszStr, g_pLeftContext->pszHost);
	// send a "done bounce" command to close the old left context
	ContextWriteString(g_pLeftContext, "done bounce");
	// send the old left host back to the caller
	ContextWriteString(p, pszStr);
	dbg_printf("wrote old left host '%s'\n", pszStr);
	// Make p the new left context and add it to the global list of contexts
	g_pLeftContext = p;
	strncpy(g_pszLeftHost, p->pszHost, MAX_HOST_LENGTH);
	g_pszLeftHost[MAX_HOST_LENGTH-1] = '\0';
	return;
    }
    else if (strnicmp(p->pszIn, "connect left ", 13) == 0)
    {
	//dbg_printf("connecting to new left host: '%s'\n", &p->pszIn[13]);
	//dbg_printf("removing left[%d]\n", p->sock);
	
	// write a "done" message to close the other end of this context
	dbg_printf("writing 'done' to close old left context.\n");
	ContextWriteString(p, "done");
	// close this context
	p->bDeleteMe = true;
	p->nState = MPD_INVALID;
	// create a new left context
	pContext = CreateContext();
	pContext->nState = MPD_IDLE;
	easy_create(&pContext->sock);
	// connect to the new left host
	dbg_printf("connecting to new left host: %s\n", &p->pszIn[13]);
	if (easy_connect(pContext->sock, &p->pszIn[13], g_nPort) == SOCKET_ERROR)
	{
	    err_printf("connect to new left host '%s' failed\n", &p->pszIn[13]);
	    RemoveContext(pContext);
	    pContext = NULL;
	    Extract(true);
	    return;
	}
	strncpy(pContext->pszHost, &p->pszIn[13], MAX_HOST_LENGTH);
	pContext->pszHost[MAX_HOST_LENGTH-1] = '\0';
	strncpy(g_pszLeftHost, &p->pszIn[13], MAX_HOST_LENGTH);
	g_pszLeftHost[MAX_HOST_LENGTH-1] = '\0';
	// authenticate with the mpd
	if (!AuthenticateConnectedConnection(&pContext))
	{
	    err_printf("HandleLeftRead: Error, authenticating new left connection to %s failed\n", &p->pszIn[13]);
	    RemoveContext(pContext);
	    pContext = NULL;
	    Extract(true);
	    return;
	}
	// indicate that this is a right connection for the remote mpd
	dbg_printf("sending 'right' to indicate a new right context.\n");
	_snprintf(pszStr, MAX_CMD_LENGTH, "right %s", g_pszHost);
	ContextWriteString(pContext, pszStr);
	// tell the remote mpd to use this context to replace its old right context
	dbg_printf("sending new right command.\n");
	ContextWriteString(pContext, "new right");
	
	pContext->nType = MPD_LEFT_SOCKET;
	pContext->nState = MPD_IDLE;
	g_pLeftContext = pContext;
	if (CreateIoCompletionPort((HANDLE)pContext->sock, g_hCommPort, (DWORD)pContext, g_NumCommPortThreads) == NULL)
	{
	    err_printf("HandleLeftRead: Unable to associate completion port with socket, error %d\n", GetLastError());
	    RemoveContext(pContext);
	    Extract(true);
	    return;
	}
	PostContextRead(pContext);
	return;
    }
    else if (strnicmp(p->pszIn, "set ", 4) == 0)
    {
	char pszKey[100];
	GetStringOpt(p->pszIn, "key", pszKey);
	GetStringOpt(p->pszIn, "value", pszStr);
	
	WriteMPDRegistry(pszKey, pszStr);
	GetStringOpt(p->pszIn, "src", pszHost);
	if ((stricmp(pszHost, g_pszHost)) && (strcmp(pszHost, g_pszIP)))
	{
	    ContextWriteString(g_pRightContext, p->pszIn);
	}
    }
    else if (strnicmp(p->pszIn, "createforwarder ", 16) == 0)
    {
	char pszBfd[10];
	GetStringOpt(p->pszIn, "sock", pszBfd);
	if (GetStringOpt(p->pszIn, "host", pszHost))
	{
	    if ((stricmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
	    {
		if (GetStringOpt(p->pszIn, "forward", pszHost))
		{
		    char *token = strtok(pszHost, ":");
		    if (token != NULL)
		    {
			token = strtok(NULL, "\n");
			int nPort = CreateIOForwarder(pszHost, atoi(token));
			
			GetStringOpt(p->pszIn, "src", pszHost);
			_snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=%d", g_pszHost, pszHost, pszBfd, nPort);
		    }
		    else
		    {
			_snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=-1", g_pszHost, pszHost, pszBfd);
		    }
		}
		else
		{
		    _snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=-1", g_pszHost, pszHost, pszBfd);
		}
		ContextWriteString(g_pRightContext, pszStr);
	    }
	    else
	    {
		GetStringOpt(p->pszIn, "src", pszHost);
		if ((stricmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
		{
		    char pszTry[10];
		    char pszForward[100];
		    if (!GetStringOpt(p->pszIn, "try", pszTry))
		    {
			GetStringOpt(p->pszIn, "forward", pszForward);
			GetStringOpt(p->pszIn, "host", pszHost);
			GetIPString(pszHost, pszHost);
			_snprintf(pszStr, MAX_CMD_LENGTH, "createforwarder src=%s host=%s sock=%s try=2 forward=%s", g_pszHost, pszHost, pszBfd, pszForward);
			ContextWriteString(g_pRightContext, pszStr);
		    }
		    else
		    {
			// command went full circle, send fail result
			_snprintf(pszStr, MAX_CMD_LENGTH, "result src=%s dest=%s sock=%s result=-1", g_pszHost, pszHost, pszBfd);
			ContextWriteString(g_pRightContext, pszStr);
		    }
		}
		else
		{
		    // forward command
		    ContextWriteString(g_pRightContext, p->pszIn);
		}
	    }