rightthread.cpp

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

		case MPD_CMD_DELETE_ID:
			strcpy(pszID, pCommand->pCommandBuffer);
			if (g_bDatabaseIsLocal)
				g_Database.Delete(pszID);
			else
			{
				pCommand->hCmd.nBufferLength = sizeof(int);
				pCommand->hCmd.cCommand = MPD_CMD_DELETE_ID;
				pCommand->hCmd.nSrcIP = nLocalIP;
				pCommand->hCmd.nSrcPort = nLocalPort;
				pBuf = pCommand->pCommandBuffer;
				n = strlen(pszID) + 1;
				*((int *)pBuf) = n;
				pBuf += sizeof(int);
				strcpy(pBuf, pszID);
				pBuf += n;
				pCommand->hCmd.nBufferLength += n;
				SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
				SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			}
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_DELETE_KEY:
			token = strtok(pCommand->pCommandBuffer, ":");
			if (token == NULL)
			{
				MarkCommandCompleted(pCommand);
				break;
			}
			strcpy(pszID, token);
			token = strtok(NULL, "\n");
			if (token == NULL)
			{
				MarkCommandCompleted(pCommand);
				break;
			}
			pszKey = new char[strlen(token)+1];
			strcpy(pszKey, token);
			if (g_bDatabaseIsLocal)
				g_Database.Delete(pszID, pszKey);
			else
			{
				pCommand->hCmd.nBufferLength = 2 * sizeof(int);
				pCommand->hCmd.cCommand = MPD_CMD_DELETE_KEY;
				pCommand->hCmd.nSrcIP = nLocalIP;
				pCommand->hCmd.nSrcPort = nLocalPort;
				pBuf = pCommand->pCommandBuffer;
				n = strlen(pszID) + 1;
				*((int *)pBuf) = n;
				pBuf += sizeof(int);
				strcpy(pBuf, pszID);
				pBuf += n;
				pCommand->hCmd.nBufferLength += n;
				n = strlen(pszKey) + 1;
				*((int *)pBuf) = n;
				pBuf += sizeof(int);
				strcpy(pBuf, pszKey);
				pBuf += n;
				pCommand->hCmd.nBufferLength += n;
				SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
				SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			}
			delete pszKey;
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_QUIT:
			break;
		case MPD_CMD_DESTROY_RING:
			pCommand->hCmd.cCommand = MPD_CMD_DESTROY_RING;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			pCommand->hCmd.nBufferLength = 0;
			pCommand->hCmd.pData = NULL;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			MarkCommandCompleted(pCommand);
			NT_closesocket(sock, sock_event);
			ExitThread(0);
			break;
		case MPD_CMD_HOSTS:
			pCommand->hCmd.cCommand = MPD_CMD_HOSTS;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			sprintf(pCommand->pCommandBuffer, "%s:%d\n", pszLocalHost, nLocalPort);
			pCommand->hCmd.nBufferLength = strlen(pCommand->pCommandBuffer);
			pCommand->hCmd.pData = pCommand;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			break;
		case MPD_CMD_CPUSAGE:
			pCommand->hCmd.cCommand = MPD_CMD_CPUSAGE;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			sprintf(pCommand->pCommandBuffer, "%s:%d %d ", pszLocalHost, nLocalPort, GetCPUsage());
			pCommand->hCmd.nBufferLength = strlen(pCommand->pCommandBuffer);
			pCommand->hCmd.pData = pCommand;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			break;
		case MPD_CMD_RUN_THE_RING:
			pCommand->hCmd.cCommand = MPD_CMD_RUN_THE_RING;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			pCommand->hCmd.pData = pCommand;
			pCommand->hCmd.nBufferLength = 0;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			break;
		case MPD_CMD_PRINT_LIST:
			g_List.Print();
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_PRINT_LISTS:
			pCommand->hCmd.cCommand = MPD_CMD_PRINT_LISTS;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			pCommand->hCmd.pData = NULL;
			pCommand->hCmd.nBufferLength = 0;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			g_List.Print();
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_PRINT_DATABASE:
			pCommand->hCmd.cCommand = MPD_CMD_PRINT_DATABASE;
			pCommand->hCmd.nBufferLength = 0;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			pCommand->hCmd.pData = pCommand;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			break;
		case MPD_CMD_LAUNCH:
			GetStringOpt(pCommand->pCommandBuffer, 'h', buffer);
			token = strtok(buffer, ":");
			if (token != NULL)
			{
				NT_get_ip(token, &nTempIP);
				/*
				{
					char *pName;
					in_addr in;
					in.S_un.S_addr = nTempIP;
					pName = inet_ntoa(in);
					printf("creating launch command for: %s\n", pName);fflush(stdout);
				}
				//*/
				token = strtok(NULL, " \n");
				if (token != NULL)
				{
					nTempPort = atoi(token);

					LaunchNode *pNode = LaunchNode::AllocLaunchNode();
					GetStringOpt(pCommand->pCommandBuffer, 'y', buffer);
					pNode->InitData((HANDLE)atoi(buffer));
					if (nTempIP == nLocalIP && nTempPort == nLocalPort)
					{
						LaunchMPDProcessArg *pArg = new LaunchMPDProcessArg;
						pArg->hEndOutput = (HANDLE)atoi(buffer);
						pArg->nIP = nLocalIP;
						pArg->nPort = nLocalPort;
						pArg->nSrcIP = nLocalIP;
						pArg->nSrcPort = nLocalPort;
						pArg->pszCommand = new char[strlen(pCommand->pCommandBuffer)+1];
						pArg->pNode = pNode;
						strcpy(pArg->pszCommand, pCommand->pCommandBuffer);
						//printf("launching '%s'\n", pArg->pszCommand);fflush(stdout);
						DWORD dwThreadID;
						CloseHandle(CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)LaunchMPDProcess, pArg, 0, &dwThreadID));
					}
					else
					{
						strcpy(buffer, pCommand->pCommandBuffer);
						pCommand->hCmd.cCommand = MPD_CMD_LAUNCH;
						pCommand->hCmd.nSrcIP = nLocalIP;
						pCommand->hCmd.nSrcPort = nLocalPort;
						pCommand->hCmd.pData = NULL;
						
						pBuf = pCommand->pCommandBuffer;
						*((LaunchNode **)pBuf) = pNode;
						pBuf = pBuf + sizeof(LaunchNode*);
						*((unsigned long *)pBuf) = nTempIP;
						pBuf = pBuf + sizeof(unsigned long);
						*((int *)pBuf) = nTempPort;
						pBuf += sizeof(int);
						strcpy(pBuf, buffer);
						
						pCommand->hCmd.nBufferLength = sizeof(LaunchNode*) + sizeof(unsigned long) + sizeof(int) + strlen(buffer)+1;
						
						SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
						SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
					}
					sprintf(pCommand->pCommandBuffer, "%d\n", pNode->GetID());
					pCommand->hCmd.nBufferLength = strlen(pCommand->pCommandBuffer) + 1;
				}
			}
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_LAUNCH_RET:
			pBuf = pCommand->pCommandBuffer;
			nTempIP = *((unsigned long *)pBuf);
			pBuf = pBuf + sizeof(unsigned long);
			nTempPort = *((int *)pBuf);
			pBuf = pBuf + sizeof(int);
			pLaunchNode = *((LaunchNode **)pBuf);
			pBuf = pBuf + sizeof(LaunchNode *);
			if (nTempIP == nLocalIP && nTempPort == nLocalPort)
			{
				pLaunchNode->Set(*((DWORD*)pBuf));
			}
			else
			{
				pCommand->hCmd.cCommand = MPD_CMD_LAUNCH_RET;
				pCommand->hCmd.nBufferLength = sizeof(unsigned long) + sizeof(int) + sizeof(LaunchNode*) + sizeof(DWORD);
				pCommand->hCmd.nSrcIP = nLocalIP;
				pCommand->hCmd.nSrcPort = nLocalPort;
				pCommand->hCmd.pData = NULL;
				
				SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
				SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			}
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_LAUNCH_EXITCODE:
			pBuf = pCommand->pCommandBuffer;
			nTempIP = *((unsigned long *)pBuf);
			pBuf = pBuf + sizeof(unsigned long);
			nTempPort = *((int *)pBuf);
			pBuf = pBuf + sizeof(int);
			pLaunchNode = *((LaunchNode **)pBuf);
			pBuf = pBuf + sizeof(LaunchNode *);
			if (nTempIP == nLocalIP && nTempPort == nLocalPort)
			{
				DWORD dwExitCode;
				int nGroup, nRank;
				dwExitCode = *((DWORD *)pBuf);
				pBuf = pBuf + sizeof(DWORD);
				nGroup = *((int *)pBuf);
				pBuf = pBuf + sizeof(int);
				nRank = *((int *)pBuf);
				pLaunchNode->SetExit(nGroup, nRank, dwExitCode);
			}
			else
			{
				pCommand->hCmd.cCommand = MPD_CMD_LAUNCH_EXITCODE;
				pCommand->hCmd.nBufferLength = sizeof(unsigned long) + sizeof(int) + sizeof(LaunchNode*) + sizeof(DWORD) + sizeof(int) + sizeof(int);
				pCommand->hCmd.nSrcIP = nLocalIP;
				pCommand->hCmd.nSrcPort = nLocalPort;
				pCommand->hCmd.pData = NULL;

				SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
				SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			}
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_KILL:
			token = strtok(pCommand->pCommandBuffer, ":");
			if (token != NULL)
			{
				NT_get_ip(token, &nTempIP);
				/*
				{
					char *pName;
					in_addr in;
					in.S_un.S_addr = nTempIP;
					pName = inet_ntoa(in);
					printf("kill: %s:", pName);fflush(stdout);
				}
				//*/
				token = strtok(NULL, " \n");
				if (token != NULL)
				{
					nTempPort = atoi(token);
					//printf("%d", nTempPort);fflush(stdout);

					token = strtok(NULL, " \n");
					if (token != NULL)
					{
						n = atoi(token);
						//printf(" (%d)\n", n);fflush(stdout);
						if (nTempIP == nLocalIP && nTempPort == nLocalPort)
							KillMPDProcess(n);
						else
						{
							strcpy(buffer, pCommand->pCommandBuffer);
							pCommand->hCmd.cCommand = MPD_CMD_KILL;
							pCommand->hCmd.nSrcIP = nLocalIP;
							pCommand->hCmd.nSrcPort = nLocalPort;
							pCommand->hCmd.pData = NULL;
							
							pBuf = pCommand->pCommandBuffer;
							*((unsigned long *)pBuf) = nTempIP;
							pBuf = pBuf + sizeof(unsigned long);
							*((int *)pBuf) = nTempPort;
							pBuf += sizeof(int);
							*((int *)pBuf) = n;
						
							pCommand->hCmd.nBufferLength = sizeof(unsigned long) + 2 * sizeof(int);
						
							SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
							SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
						}
					}
				}
			}
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_KILL_GROUP:
			n = atoi(pCommand->pCommandBuffer);
			//printf("killing group %d\n", n);fflush(stdout);

			strcpy(buffer, pCommand->pCommandBuffer);
			pCommand->hCmd.cCommand = MPD_CMD_KILL_GROUP;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			pCommand->hCmd.pData = NULL;
							
			pBuf = pCommand->pCommandBuffer;
			*((int *)pBuf) = n;

			pCommand->hCmd.nBufferLength = sizeof(int);

			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);

			KillMPDProcesses(n);
			MarkCommandCompleted(pCommand);
			break;
		case MPD_CMD_PS:
			pCommand->hCmd.cCommand = MPD_CMD_PS;
			pCommand->hCmd.nSrcIP = nLocalIP;
			pCommand->hCmd.nSrcPort = nLocalPort;
			sprintf(pShortBuffer, "%s:%d", pszLocalHost, nLocalPort);
			PrintMPDProcessesToBuffer(pCommand->pCommandBuffer, pShortBuffer);
			pCommand->hCmd.nBufferLength = strlen(pCommand->pCommandBuffer);
			pCommand->hCmd.pData = pCommand;
			SendBlocking(sock, (char*)&pCommand->hCmd, sizeof(CommandHeader), 0);
			SendBlocking(sock, pCommand->pCommandBuffer, pCommand->hCmd.nBufferLength, 0);
			break;
		default:
			pCommand->bSuccess = false;
			sprintf(pCommand->pCommandBuffer, "Unknown command\n");
			pCommand->hCmd.nBufferLength = strlen(pCommand->pCommandBuffer)+1;
			MarkCommandCompleted(pCommand);
		}
	}

	NT_closesocket(sock, sock_event);
}