nt_ipvishm_control_loop.cpp

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

			cArg->sock_event = temp_event;
			hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ControlLoopClientThread, cArg, NT_THREAD_STACK_SIZE, &dwThreadID);
			if (hThread == NULL)
			{
				delete cArg;
				NT_Tcp_closesocket(temp_socket, temp_event);
				nt_error_socket("CreateThread failed in ControlLoopThread.", GetLastError());
			}
			CloseHandle(hThread);
			continue;
		}
		result = GetLastError();
		if (result == WSAEWOULDBLOCK)
		{
			WSAResetEvent(sock_event);
			WSAEventSelect(sock, sock_event, FD_ACCEPT);
		}
		else
			nt_error_socket("ControlLoopThread: accept failed", result);
	}
}

// Function name	: SendInitDataToRoot
// Description	    : 
// Return type		: bool 
bool SendInitDataToRoot()
{
	int ret_val;
	char ack, cmd;
	WSAEVENT sock_event;
	SOCKET sock;

	// create the event
	sock_event = WSACreateEvent();
	if (sock_event == WSA_INVALID_EVENT)
		nt_error_socket("WSACreateEvent failed in SendInitDataToRoot", WSAGetLastError());
	// create the socket
	sock = socket(PF_INET, SOCK_STREAM, 0);
	if (sock == INVALID_SOCKET)
		nt_error_socket("socket failed in SendInitDataToRoot", WSAGetLastError());

	//printf("Connecting to root: %s %d\n", g_pszRootHostName, g_nRootPort);fflush(stdout);
	ret_val = NT_Tcp_connect(sock, g_pszRootHostName, g_nRootPort);
	if (ret_val)
		nt_error_socket("SendInitDataToRoot: NT_Tcp_connect failed", ret_val);

	if (WSAEventSelect(sock, sock_event, FD_READ | FD_CLOSE) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: WSAEventSelect failed", WSAGetLastError());

	cmd = NT_TCP_CTRL_CMD_INIT_DATA_TO_ROOT;
	if (SendBlocking(sock, &cmd, 1, 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send cmd failed", WSAGetLastError());

	// Send iproc, listen_port, hostname, exename, and pid
	if (SendBlocking(sock, (char*)&g_nIproc, sizeof(int), 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send iproc failed", WSAGetLastError());
	if (SendBlocking(sock, (char*)&g_pProcTable[g_nIproc].listen_port, sizeof(int), 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send listen port failed", WSAGetLastError());
	if (SendBlocking(sock, (char*)&g_pProcTable[g_nIproc].control_port, sizeof(int), 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send control port failed", WSAGetLastError());
	if (SendBlocking(sock, g_pszHostName, NT_HOSTNAME_LEN, 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send host name failed", WSAGetLastError());
	if (SendBlocking(sock, g_pProcTable[g_nIproc].exename, NT_EXENAME_LEN, 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send exe name failed", WSAGetLastError());
	if (SendBlocking(sock, (char*)&g_pProcTable[g_nIproc].pid, sizeof(int), 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send pid failed", WSAGetLastError());
	if (SendBlocking(sock, (char*)&g_pProcTable[g_nIproc].num_nics, sizeof(int), 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send num_nics failed", WSAGetLastError());
	if (SendBlocking(sock, (char*)&g_pProcTable[g_nIproc].nic_ip, sizeof(int)*MAX_NUM_NICS, 0) == SOCKET_ERROR)
		nt_error_socket("SendInitDataToRoot: send nic_ip[4] failed", WSAGetLastError());

	// Wait for an ack to ensure the data was received
	ret_val = ReceiveBlocking(sock, sock_event, &ack, 1, 0);
	if (ret_val)
		nt_error_socket("SendInitDataToRoot: recv ack failed", WSAGetLastError());

	//printf("SendInitDataToRoot called.\n");fflush(stdout);

	NT_Tcp_closesocket(sock, sock_event);
	return true;
}

// Function name	: GetProcessConnectInfo
// Description	    : 
// Return type		: bool 
// Argument         : int iproc
bool GetProcessConnectInfo(int iproc)
{
	int ret_val;
	char cmd;
	WSAEVENT sock_event;
	SOCKET sock;

	// create the event
	sock_event = WSACreateEvent();
	if (sock_event == WSA_INVALID_EVENT)
		nt_error_socket("WSACreateEvent failed in GetProcessConnectInfo", WSAGetLastError());
	// create the socket
	sock = socket(PF_INET, SOCK_STREAM, 0);
	if (sock == INVALID_SOCKET)
		nt_error_socket("socket failed in GetProcessConnectInfo", WSAGetLastError());

	//printf("Connecting to root: %s %d\n", host, port);fflush(stdout);
	ret_val = NT_Tcp_connect(sock, g_pszRootHostName, g_nRootPort);
	if (ret_val)
		nt_error_socket("GetProcessConnectInfo: NT_Tcp_connect failed", ret_val);

	if (WSAEventSelect(sock, sock_event, FD_READ | FD_CLOSE) == SOCKET_ERROR)
		nt_error_socket("GetProcessConnectInfo: WSAEventSelect failed", WSAGetLastError());

	cmd = NT_TCP_CTRL_CMD_PROCESS_CONNECT_INFO;
	if (SendBlocking(sock, &cmd, 1, 0) == SOCKET_ERROR)
		nt_error_socket("GetProcessConnectInfo: send cmd failed", WSAGetLastError());

	// Send the rank of the process information is requested of
	if (SendBlocking(sock, (char*)&iproc, sizeof(int), 0) == SOCKET_ERROR)
		MakeErrMsg(WSAGetLastError(), "GetProcessConnectInfo: send iproc(%d) to root failed", iproc);
	// Receive the host name and port for the requested process
	if (ret_val = ReceiveBlocking(sock, sock_event, g_pProcTable[iproc].host, NT_HOSTNAME_LEN, 0))
		MakeErrMsg(ret_val, "GetProcessConnectInfo: receive host name %d failed", iproc);
	if (ret_val = ReceiveBlocking(sock, sock_event, (char*)&g_pProcTable[iproc].listen_port, sizeof(int), 0))
		MakeErrMsg(ret_val, "GetProcessConnectInfo: receive listen_port %d failed", iproc);

	//printf("GetProcessConnectInfo called\n");fflush(stdout);
	NT_Tcp_closesocket(sock, sock_event);

	if (g_pProcTable[iproc].listen_port < 1)
		return false;
	return true;
}

// Function name	: GetProcessInfo
// Description	    : 
// Return type		: bool 
// Argument         : int iproc
bool GetProcessInfo(int iproc)
{
	int ret_val;
	char cmd;
	WSAEVENT sock_event;
	SOCKET sock;

	// create the event
	sock_event = WSACreateEvent();
	if (sock_event == WSA_INVALID_EVENT)
		nt_error_socket("WSACreateEvent failed in GetProcessInfo", WSAGetLastError());
	// create the socket
	sock = socket(PF_INET, SOCK_STREAM, 0);
	if (sock == INVALID_SOCKET)
		nt_error_socket("socket failed in GetProcessInfo", WSAGetLastError());

	//printf("Connecting to root: %s %d\n", host, port);fflush(stdout);
	ret_val = NT_Tcp_connect(sock, g_pszRootHostName, g_nRootPort);
	if (ret_val)
		nt_error_socket("GetProcessInfo: NT_Tcp_connect failed", ret_val);

	if (WSAEventSelect(sock, sock_event, FD_READ | FD_CLOSE) == SOCKET_ERROR)
		nt_error_socket("GetProcessInfo: WSAEventSelect failed", WSAGetLastError());

	cmd = NT_TCP_CTRL_CMD_PROCESS_INFO;
	if (SendBlocking(sock, &cmd, 1, 0) == SOCKET_ERROR)
		nt_error_socket("GetProcessInfo: send cmd failed", WSAGetLastError());

	// Send the rank of the process information is requested of
	if (SendBlocking(sock, (char*)&iproc, sizeof(int), 0) == SOCKET_ERROR)
		MakeErrMsg(WSAGetLastError(), "GetProcessInfo: SendBlocking iproc(%d) failed", iproc);
	// Receive the host name, executable name, and process id for the requested process
	if (ret_val = ReceiveBlocking(sock, sock_event, g_pProcTable[iproc].host, NT_HOSTNAME_LEN, 0))
		MakeErrMsg(ret_val, "GetProcessInfo: receive host %d failed", iproc);
	if (ret_val = ReceiveBlocking(sock, sock_event, g_pProcTable[iproc].exename, NT_EXENAME_LEN, 0))
		MakeErrMsg(ret_val, "GetProcessInfo: receive exename %d failed", iproc);
	if (ret_val = ReceiveBlocking(sock, sock_event, (char*)&g_pProcTable[iproc].pid, sizeof(int), 0))
		MakeErrMsg(ret_val, "GetProcessInfo: receive process %d pid failed", iproc);

	//printf("GetProcessInfo called\n");fflush(stdout);
	NT_Tcp_closesocket(sock, sock_event);
	return true;
}

// Function name	: SendInDoneMsg
// Description	    : 
// Return type		: bool 
bool SendInDoneMsg()
{
	int ret_val;
	char ack, cmd;
	WSAEVENT sock_event;
	SOCKET sock;

	// create the event
	sock_event = WSACreateEvent();
	if (sock_event == WSA_INVALID_EVENT)
		nt_error_socket("WSACreateEvent failed in SendInDoneMsg", WSAGetLastError());
	// create the socket
	sock = socket(PF_INET, SOCK_STREAM, 0);
	if (sock == INVALID_SOCKET)
		nt_error_socket("socket failed in SendInDoneMsg", WSAGetLastError());

	ret_val = NT_Tcp_connect(sock, g_pszRootHostName, g_nRootPort);
	if (ret_val)
		nt_error_socket("SendInDoneMsg: NT_Tcp_connect failed", ret_val);

	if (WSAEventSelect(sock, sock_event, FD_READ | FD_CLOSE) == SOCKET_ERROR)
		nt_error_socket("SendInDoneMsg: WSAEventSelect failed", WSAGetLastError());

	cmd = NT_TCP_CTRL_CMD_POST_IN_DONE;
	if (SendBlocking(sock, &cmd, 1, 0) == SOCKET_ERROR)
		nt_error_socket("SendInDoneMsg: send cmd failed", WSAGetLastError());

	ret_val = ReceiveBlocking(sock, sock_event, &ack, 1, 0);
	if (ret_val)
		nt_error_socket("SendInDoneMsg: receive ack failed", ret_val);

	NT_Tcp_closesocket(sock, sock_event);
	return true;
}

// Function name	: SendAllDoneMsg
// Description	    : 
// Return type		: bool 
// Argument         : char *host
// Argument         : int port
bool SendAllDoneMsg(char *host, int port)
{
	int ret_val;
	char ack, cmd;
	WSAEVENT sock_event;
	SOCKET sock;

	// create the event
	sock_event = WSACreateEvent();
	if (sock_event == WSA_INVALID_EVENT)
		nt_error_socket("WSACreateEvent failed in SendAllDoneMsg", WSAGetLastError());
	// create the socket
	sock = socket(PF_INET, SOCK_STREAM, 0);
	if (sock == INVALID_SOCKET)
		nt_error_socket("socket failed in SendAllDoneMsg", WSAGetLastError());

	ret_val = NT_Tcp_connect(sock, host, port);
	if (ret_val)
		nt_error_socket("SendAllDoneMsg: NT_Tcp_connect failed", ret_val);

	if (WSAEventSelect(sock, sock_event, FD_READ | FD_CLOSE) == SOCKET_ERROR)
		nt_error_socket("SendAllDoneMsg: WSAEventSelect failed", WSAGetLastError());

	cmd = NT_TCP_CTRL_CMD_ALL_IN_DONE;
	if (SendBlocking(sock, &cmd, 1, 0) == SOCKET_ERROR)
		nt_error_socket("SendAllDoneMsg: send cmd failed", WSAGetLastError());

	ret_val = ReceiveBlocking(sock, sock_event, &ack, 1, 0);
	if (ret_val)
		nt_error_socket("SendAllDoneMsg: receive ack failed", ret_val);

	NT_Tcp_closesocket(sock, sock_event);
	return true;
}