mpdconsole.cpp

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

	WriteString(sock, "ERROR - no file name provided");
	return;
    }

    // Open the file
    fin = fopen(pszFileName, "rb");
    if (fin == NULL)
    {
	nError = GetLastError();
	Translate_Error(nError, pszStr, "ERROR - fopen failed, ");
	WriteString(sock, "-1");
	WriteString(sock, pszStr);
	return;
    }

    // Send the size
    fseek(fin, 0, SEEK_END);
    nLength = ftell(fin);
    if (nLength == -1)
    {
	nError = GetLastError();
	Translate_Error(nError, pszStr, "ERROR - Unable to determine the size of the file, ");
	WriteString(sock, "-1");
	WriteString(sock, pszStr);
	return;
    }
    sprintf(pszStr, "%d", nLength);
    WriteString(sock, pszStr);

    // Rewind back to the beginning
    fseek(fin, 0, SEEK_SET);

    // Send the data
    while (nLength)
    {
	nNumRead = min(nLength, TRANSFER_BUFFER_SIZE);
	nNumRead = fread(pBuffer, 1, nNumRead, fin);
	if (nNumRead < 1)
	{
	    err_printf("fread failed, %d\n", ferror(fin));
	    fclose(fin);
	    return;
	}
	if (easy_send(sock, pBuffer, nNumRead) == SOCKET_ERROR)
	{
	    err_printf("sending file data failed, file=%s, error=%d", pszFileName, WSAGetLastError());
	    fclose(fin);
	    return;
	}
	//dbg_printf("%d bytes sent\n", nNumRead);fflush(stdout);
	nLength -= nNumRead;
    }
    fclose(fin);
}

static void ConsoleGetFile(SOCKET sock, char *pszInputStr)
{
    bool bReplace = true, bCreateDir = false;
    char pszFileName[MAX_PATH];
    char pszStr[256];
    int nLength;
    FILE *fout;
    char pBuffer[TRANSFER_BUFFER_SIZE];
    int nNumRead;
    int nNumWritten;

    if (GetStringOpt(pszInputStr, "replace", pszStr))
    {
	bReplace = (stricmp(pszStr, "yes") == 0);
    }
    if (GetStringOpt(pszInputStr, "createdir", pszStr))
    {
	bCreateDir = (stricmp(pszStr, "yes") == 0);
    }
    if (GetStringOpt(pszInputStr, "length", pszStr))
    {
	nLength = atoi(pszStr);
	//dbg_printf("nLength: %d\n", nLength);
    }
    else
    {
	WriteString(sock, "ERROR - length not provided");
	return;
    }
    if (nLength < 1)
    {
	WriteString(sock, "ERROR - invalid length");
	return;
    }

    if (!GetStringOpt(pszInputStr, "name", pszFileName))
    {
	WriteString(sock, "ERROR - no file name provided");
	return;
    }

    //dbg_printf("creating file '%s'\n", pszFileName);
    fout = CreateCheckFile(pszFileName, bReplace, bCreateDir, pszStr);

    if (fout == NULL)
    {
	WriteString(sock, pszStr);
	return;
    }

    //dbg_printf("SEND\n");
    WriteString(sock, "SEND");

    while (nLength)
    {
	nNumRead = min(nLength, TRANSFER_BUFFER_SIZE);
	if (easy_receive(sock, pBuffer, nNumRead) == SOCKET_ERROR)
	{
	    err_printf("ERROR: easy_receive failed, error %d\n", WSAGetLastError());
	    fclose(fout);
	    DeleteFile(pszFileName);
	    return;
	}
	nNumWritten = fwrite(pBuffer, 1, nNumRead, fout);
	if (nNumWritten != nNumRead)
	{
	    err_printf("ERROR: received %d bytes but only wrote %d bytes\n", nNumRead, nNumWritten);
	}
	//dbg_printf("%d bytes read, %d bytes written\n", nNumRead, nNumWritten);
	nLength -= nNumRead;
    }

    fclose(fout);

    WriteString(sock, "SUCCESS");
}

static void GetDirectoryFiles(SOCKET sock, char *pszInputStr)
{
    char pszPath[MAX_PATH];
    char pszStr[MAX_CMD_LENGTH];
    int nFolders = 0, nFiles = 0;
    WIN32_FIND_DATA data;
    HANDLE hFind;

    if (!GetStringOpt(pszInputStr, "path", pszPath))
    {
	WriteString(sock, "ERROR: no path specified");
	return;
    }
    if (strlen(pszPath) < 1)
    {
	WriteString(sock, "ERROR: empty path specified");
	return;
    }

    if (pszPath[strlen(pszPath)-1] != '\\')
    {
	strcat(pszPath, "\\");
    }
    strcat(pszPath, "*");

    // Count the files and folders
    // What if the contents change between the counting and the sending?
    hFind = FindFirstFile(pszPath, &data);

    if (hFind == INVALID_HANDLE_VALUE)
    {
	Translate_Error(GetLastError(), pszStr, "ERROR: ");
	WriteString(sock, pszStr);
	return;
    }

    if (data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
    {
	if (strcmp(data.cFileName, ".") && strcmp(data.cFileName, ".."))
	    nFolders++;
    }
    else
	nFiles++;

    while (FindNextFile(hFind, &data))
    {
	if (data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
	{
	    if (strcmp(data.cFileName, ".") && strcmp(data.cFileName, ".."))
		nFolders++;
	}
	else
	    nFiles++;
    }

    FindClose(hFind);

    // Send the folders
    sprintf(pszStr, "%d", nFolders);
    WriteString(sock, pszStr);

    hFind = FindFirstFile(pszPath, &data);

    if (hFind == INVALID_HANDLE_VALUE)
	return;

    if (data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
    {
	if (strcmp(data.cFileName, ".") && strcmp(data.cFileName, ".."))
	    WriteString(sock, data.cFileName);
    }

    while (FindNextFile(hFind, &data))
    {
	if (data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
	{
	    if (strcmp(data.cFileName, ".") && strcmp(data.cFileName, ".."))
		WriteString(sock, data.cFileName);
	}
    }

    FindClose(hFind);

    // Send the files
    sprintf(pszStr, "%d", nFiles);
    WriteString(sock, pszStr);

    hFind = FindFirstFile(pszPath, &data);

    if (hFind == INVALID_HANDLE_VALUE)
	return;

    if (!(data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY))
    {
	WriteString(sock, data.cFileName);
	if (data.nFileSizeHigh > 0)
	    sprintf(pszStr, "%d:%d", data.nFileSizeLow, data.nFileSizeHigh);
	else
	    sprintf(pszStr, "%d", data.nFileSizeLow);
	WriteString(sock, pszStr);
    }

    while (FindNextFile(hFind, &data))
    {
	if (!(data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY))
	{
	    WriteString(sock, data.cFileName);
	    if (data.nFileSizeHigh > 0)
		sprintf(pszStr, "%d:%d", data.nFileSizeLow, data.nFileSizeHigh);
	    else
		sprintf(pszStr, "%d", data.nFileSizeLow);
	    WriteString(sock, pszStr);
	}
    }

    FindClose(hFind);
}

static void HandleDBCommandRead(MPD_Context *p)
{
    char name[MAX_DBS_NAME_LEN+1] = "";
    char key[MAX_DBS_KEY_LEN+1] = "";
    char value[MAX_DBS_VALUE_LEN+1] = "";
    char pszStr[MAX_CMD_LENGTH] = "";

    if (strnicmp(p->pszIn, "dbput ", 6) == 0)
    {
	GetNameKeyValue(&p->pszIn[6], name, key, value);
	if (dbs_put(name, key, value) == DBS_SUCCESS)
	    ContextWriteString(p, DBS_SUCCESS_STR);
	else
	    ContextWriteString(p, DBS_FAIL_STR);
    }
    else if (strnicmp(p->pszIn, "dbget ", 6) == 0)
    {
	GetNameKeyValue(&p->pszIn[6], name, key, NULL);
	if (dbs_get(name, key, value) == DBS_SUCCESS)
	    ContextWriteString(p, value);
	else
	{
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbget src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[6]);
	    ContextWriteString(g_pRightContext, pszStr);
	}
    }
    else if (stricmp(p->pszIn, "dbcreate") == 0)
    {
	// Create the database locally
	if (dbs_create(name) == DBS_SUCCESS)
	{
	    // Write the name back to the user
	    ContextWriteString(p, name);
	    // Create the database on all the other nodes
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbcreate src=%s sock=%d name=%s", g_pszHost, p->sock, name);
	    ContextWriteString(g_pRightContext, pszStr);
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (strnicmp(p->pszIn, "dbcreate ", 9) == 0)
    {
	GetNameKeyValue(&p->pszIn[9], name, NULL, NULL);
	if (dbs_create_name_in(name) == DBS_SUCCESS)
	{
	    ContextWriteString(p, DBS_SUCCESS_STR);
	    // Create the database on all the other nodes
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbcreate src=%s sock=%d name=%s", g_pszHost, p->sock, name);
	    ContextWriteString(g_pRightContext, pszStr);
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (strnicmp(p->pszIn, "dbdestroy ", 10) == 0)
    {
	// forward the destroy command
	_snprintf(pszStr, MAX_CMD_LENGTH, "dbdestroy src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[10]);
	ContextWriteString(g_pRightContext, pszStr);

	// destroy the database locally
	GetNameKeyValue(&p->pszIn[10], name, NULL, NULL);
	if (dbs_destroy(name) == DBS_SUCCESS)
	    ContextWriteString(p, DBS_SUCCESS_STR);
	else
	    ContextWriteString(p, DBS_FAIL_STR);
    }
    else if (strnicmp(p->pszIn, "dbfirst ", 8) == 0)
    {
	GetNameKeyValue(&p->pszIn[8], name, NULL, NULL);
	if (dbs_first(name, key, value) == DBS_SUCCESS)
	{
	    // forward the first command
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbfirst src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[8]);
	    ContextWriteString(g_pRightContext, pszStr);

	    if (*key == '\0')
	    {
		// If the local database is empty, forward a dbnext command
		_snprintf(pszStr, MAX_CMD_LENGTH, "dbnext src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[8]);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	    else
	    {
		_snprintf(p->pszOut, MAX_CMD_LENGTH, "key=%s value=%s", key, value);
		ContextWriteString(p, p->pszOut);
	    }
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (strnicmp(p->pszIn, "dbnext ", 7) == 0)
    {
	GetNameKeyValue(&p->pszIn[7], name, NULL, NULL);
	if (dbs_next(name, key, value) == DBS_SUCCESS)
	{
	    if (*key == '\0')
	    {
		_snprintf(pszStr, MAX_CMD_LENGTH, "dbnext src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[7]);
		ContextWriteString(g_pRightContext, pszStr);
	    }
	    else
	    {
		_snprintf(p->pszOut, MAX_CMD_LENGTH, "key=%s value=%s", key, value);
		ContextWriteString(p, p->pszOut);
	    }
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (stricmp(p->pszIn, "dbfirstdb") == 0)
    {
	if (dbs_firstdb(name) == DBS_SUCCESS)
	{
	    if (*name == '\0')
		strcpy(p->pszOut, DBS_END_STR);
	    else
		_snprintf(p->pszOut, MAX_CMD_LENGTH, "name=%s", name);
	    ContextWriteString(p, p->pszOut);
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (stricmp(p->pszIn, "dbnextdb") == 0)
    {
	if (dbs_nextdb(name) == DBS_SUCCESS)
	{
	    if (*name == '\0')
		strcpy(p->pszOut, DBS_END_STR);
	    else
		_snprintf(p->pszOut, MAX_CMD_LENGTH, "name=%s", name);
	    ContextWriteString(p, p->pszOut);
	}
	else
	{
	    ContextWriteString(p, DBS_FAIL_STR);
	}
    }
    else if (strnicmp(p->pszIn, "dbdelete ", 9) == 0)
    {
	// Attempt to delete locally
	GetNameKeyValue(&p->pszIn[9], name, key, NULL);
	if (dbs_delete(name, key) == DBS_SUCCESS)
	{
	    ContextWriteString(p, DBS_SUCCESS_STR);
	}
	else
	{
	    // forward the delete command
	    _snprintf(pszStr, MAX_CMD_LENGTH, "dbdelete src=%s sock=%d %s", g_pszHost, p->sock, &p->pszIn[9]);
	    ContextWriteString(g_pRightContext, pszStr);
	}
    }
    else
    {
	err_printf("unknown command '%s'", p->pszIn);
    }
}

void statConfig(char *pszOutput, int length)
{
    pszOutput[0] = '\0';
    MPDRegistryToString(pszOutput, length);
}

void HandleConsoleRead(MPD_Context *p)
{
    char pszStr[MAX_CMD_LENGTH];

    dbg_printf("ConsoleRead[%d]: '%s'\n", p->sock, p->pszIn);
    switch(p->nLLState)
    {
    case MPD_READING_CMD:
	if (strnicmp(p->pszIn, "db", 2) == 0)
	{
	    HandleDBCommandRead(p);
	}
	else if (strnicmp(p->pszIn, "launch ", 7) == 0)
	{
	    char pszHost[MAX_HOST_LENGTH];
	    g_nCurrentLaunchId++;
	    LaunchStateStruct *pLS = new LaunchStateStruct;
	    pLS->nStatus = LAUNCH_PENDING;
	    strcpy(pLS->pszError, "LAUNCH_PENDING");
	    pLS->nId = g_nCurrentLaunchId;
	    pLS->nBfd = p->sock;
	    pLS->pNext = g_pLaunchList;
	    if (!GetStringOpt(&p->pszIn[7], "h", pLS->pszHost))
	    {
		strncpy(pLS->pszHost, g_pszHost, MAX_HOST_LENGTH);
		pLS->pszHost[MAX_HOST_LENGTH-1] = '\0';
	    }
	    g_pLaunchList = pLS;
	    _snprintf(pszStr, MAX_CMD_LENGTH, "launch src=%s id=%d %s", g_pszHost, pLS->nId, &p->pszIn[7]);
	    if (GetStringOpt(pszStr, "h", pszHost))
	    {
		if ((stricmp(pszHost, g_pszHost) == 0) || (strcmp(pszHost, g_pszIP) == 0))
		    Launch(pszStr);
		else
		    ContextWriteString(g_pRightContext, pszStr);
	    }
	    else
		Launch(pszStr); // No host provided so launch locally
	    sprintf(pszStr, "%d", pLS->nId);
	    ContextWriteString(p, pszStr);
	}
	else if (strnicmp(p->pszIn, "getpid ", 7) == 0)
	{
	    LaunchStateStruct *pLS = GetLaunchStruct(atoi(&p->pszIn[7]));
	    if (pLS != NULL)
	    {
		WaitForSingleObject(pLS->hMutex, INFINITE);
		if (pLS->nStatus == LAUNCH_PENDING)
		{
		    pLS->bPidRequested = true;
		    ReleaseMutex(pLS->hMutex);
		}
		else
		{
		    if (pLS->nStatus == LAUNCH_SUCCESS)
			sprintf(pszStr, "%d", pLS->nPid);
		    else
			strcpy(pszStr, "-1");
		    ReleaseMutex(pLS->hMutex);
		    ContextWriteString(p, pszStr);
		}
	    }
	    else
	    {
		ContextWriteString(p, "-1");
	    }
	}
	else if (strnicmp(p->pszIn, "getexitcode ", 12) == 0)
	{
	    LaunchStateStruct *pLS = GetLaunchStruct(atoi(&p->pszIn[12]));
	    if (pLS != NULL)
	    {
		WaitForSingleObject(pLS->hMutex, INFINITE);
		if (pLS->nStatus == LAUNCH_EXITED)
		{
		    dbg_printf("HandleConsoleRead:Sending exit code %d for launchid %d\n", pLS->nExitCode, atoi(&p->pszIn[12]));
		    sprintf(pszStr, "%d", pLS->nExitCode);
		}
		else
		{
		    if (pLS->nStatus == LAUNCH_SUCCESS)
			strcpy(pszStr, "ACTIVE");
		    else
			strcpy(pszStr, "FAIL");
		}
		ReleaseMutex(pLS->hMutex);
		ContextWriteString(p, pszStr);
	    }
	    else
	    {
		ContextWriteString(p, "FAIL");
	    }
	}
	else if (strnicmp(p->pszIn, "getexitcodewait ", 16) == 0)
	{
	    LaunchStateStruct *pLS = GetLaunchStruct(atoi(&p->pszIn[16]));
	    if (pLS != NULL)
	    {
		WaitForSingleObject(pLS->hMutex, INFINITE);
		if (pLS->nStatus == LAUNCH_SUCCESS)
		{
		    pLS->bExitStateRequested = true;
		    ReleaseMutex(pLS->hMutex);
		}
		else
		{
		    if (pLS->nStatus == LAUNCH_EXITED)
		    {
			dbg_printf("sending exit code %d:%d\n", atoi(&p->pszIn[16]), pLS->nExitCode);
			sprintf(pszStr, "%d", pLS->nExitCode);
		    }
		    else
			strcpy(pszStr, "FAIL");
		    ReleaseMutex(pLS->hMutex);
		    ContextWriteString(p, pszStr);
		}
	    }
	    else
	    {
		ContextWriteString(p, "FAIL");
	    }
	}
	else if (strnicmp(p->pszIn, "getexittime ", 12) == 0)
	{
	    LaunchStateStruct *pLS = GetLaunchStruct(atoi(&p->pszIn[12]));
	    if (pLS != NULL)
	    {
		if (strlen(pLS->timestamp))
		{
		    dbg_printf("sending exit time %d:%s\n", atoi(&p->pszIn[12]), pLS->timestamp);
		    strcpy(pszStr, pLS->timestamp);
		}
		else
		{
		    if (pLS->nStatus == LAUNCH_SUCCESS)
		    {
			strcpy(pszStr, "ACTIVE");
		    }
		    else
		    {
			if (pLS->timestamp[0] == '\0')
			{
			    strcpy(pszStr, "unknown");
			}
			else
			{
			    dbg_printf("sending exit time %d:%s\n", atoi(&p->pszIn[12]), pLS->timestamp);
			    strcpy(pszStr, pLS->timestamp);
			}
		    }
		}
		ContextWriteString(p, pszStr);
	    }
	    else
	    {
		ContextWriteString(p, "FAIL");
	    }
	}
	else if (strnicmp(p->pszIn, "getmpifinalized ", 16) == 0)