[TCP/IP 소켓 프로그래밍] (18) IOCP 서버

출처 : 열혈 TCP/IP 소켓 프로그래밍 윤성우 저

 

IOCP 서버

Overlapped IO는 입출력 자체를 중첩하여 수행하는 의미이다.

IOCP에서는 중첩뿐만아니라 입출력이 완료된 이후 처리를 별도의 쓰레드가 담당하도록 해준다.

 

Completion Port 오브젝트

IOCP에서 완료된 IO의 정보가 Completion Port 오브젝트라는 커널 오브젝트에 등록된다. 어떤 IO의 완료 상황을 처리할 것인가 그것을 트리거하는 주체는 누가될 것인가를 정해야한다. 그 주체는 바로 epoll과 같이 소켓 핸들이며 CP 오브젝트와 연결하여 사용한다.

 

HANDLE CreateIoCompletionPort(HANDLE FileHandle, HANDLE ExistingCompletionPort,
					ULONG_PTR CompletionKey, DWORD NumberOfConcurrentThreads);

• FileHandle: CP 오브젝트와 연결할 소켓 핸들
• ExistingCompletionPort: 소켓과 연결할 CP 오브젝트 핸들
• CompletionKey: 완료된 IO관련 정보의 전달을 위한 매개변수
• NumberOfConcurrentThreads: CP 오브젝트에 할당되어 완료된 IO를 처리할 쓰레드 수를 전달

위 함수는 CP 오브젝트를 생성할 때와 소켓 핸들과 연결할 때 다른 방식으로 사용한다.

CP 오브젝트 생성

CP 오브젝트 생성시에는 쓰레드 수를 결정하는 마지막 인자만 사용한다.

HANDLE hCpObject;
hCpObject = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 2);

 

CP 오브젝트 연결

HANDLE hCpObject;
SOCKET hSock;

CreateIoCompletionPort((HANDLE)hSock, hCpObject, (DWORD)ioInfo, 0);

생성한 소켓과 CP 오브젝트를 함수 인자로 전달하면 두 오브젝트가 연결된다.

 

Completion Port의 완료된 IO를 확인하고 처리를 위한 쓰레드에서 사용되는 함수이다.

BOOL GetQueuedCompletionStatus(HANDLE CompletionPort, LPDWORD lpNumberOfBytes,
						PULONG_PTR lpCompletionKey, LPOVERLAPPED* lpOverlapped, WORD dwmilliseconds);

• CompletionPort: 완료된 IO 정보가 등록되어 이쓴 CP 오브젝트 핸들 전달
• lpNumberOfBytes: 송수신 된 데이터의 크기정보를 전달하기 위한 변수의 주소값
• lpCompletionKey: CreateIOCompletionPort에 전달된 CompletionKey를 얻기 위한 주소값
• lpOverlapped: WSASend와 WSARecv 함수 호출 시 전달하는 OVERLAPPED 구조체 변수의 주소 값이 저장될 변수의 주소값 전달
• dwMilliseconds: 타임아웃 정보전달, 지정한 시간이 완료되면 FALSE를 반환하면서 함수 반환, INFINITE시 완료된 IO가 CP오브젝트에 등록될때까지 블로킹 상태

GetQueueCompletionStatus를 통해 얻고자 하는 값은 IO가 완료된 이후의 상태에서 호출이 되기 때문에 이후 서비스를 제공하기 위한 "메세지 버퍼"와 "소켓" 정보를 lpCompletionKey와 lpOverlapped를 통해 얻어와야한다.

 

IOCP 에코 서버 구현

"메세지 버퍼"와 서비스를 제공할 "소켓" 정보를 담기 위한 사용자 구조체를 정의한다.

typedef struct // sockt info
{
	SOCKET hClntSock;
	SOCKADDR_IN clntAdr;
} PER_HANLDE_DATA, *LPPER_HANDLE_DATA;

typedef struct
{
	OVERLAPPED overlapped;
	WSABUF wsaBuf;
	char buffer[BUF_SIZE];
	int rwMode; // read or write
} PER_IO_DATA, *LPPER_IO_DATA;

PER_HANDLE_DATA는 클라이언트 소켓과 주소 정보를 담을 예정이고 PER_IO_DATA는 OVERLAPPED 구조체가 첫번째 멤버로 설정되어 있기 때문에 "메세지 버퍼"용 데이터를 정의한다.

 

main 함수

main 함수에서 처리되는 로직은 다음과 같다.

1. IO Completion Port 객체 (CP 오브젝트)를 생성하하고 CP 오브젝트를 인수로 하는 IO 완료후 서비스를 제공하는 쓰레드를 실행한다. 이 때 아직 완료된 IO는 없기 때문에 GetQueueCompletionStatus는 블로킹 상태가 된다.

2. 기본 Overlapped IO 소켓을 생성하고 주소 할당 및 소켓을 연결 대기 상태에 놓는다.

3. 소켓에 클라이언트 요청이 들어오면 accept 함수는 클라이언트 핸들을 반환한다.

4. "소켓" 정보를 담는 구조체를 동적 할당하고 연결된 클라이언트 정보로 채운다.

5. 클라이언트 소켓과 CP 오브젝트를 연결하고 CompletionKey 인자에 "소켓" 정보 구조체를 전달한다.

6. "메세지 버퍼"를 담는 구조체를 동적 할당하고 WSARecv 함수를 호출하여 중첩 입력 동작을 수행한다.

DWORD WINAPI EchoThreadMain(LPVOID CompletionPortIO);

int main(int argc, char* argv[])
{

	HANDLE hComPort;
	SYSTEM_INFO sysInfo;
	LPPER_IO_DATA ioInfo;
	LPPER_HANDLE_DATA handleInfo;

	SOCKET hServSock;
	SOCKADDR_IN servAdr;
	int recvBytes, flags = 0;

	// 윈속 라이브러리 초기화
	WSADATA wsaData;
	if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
		ErrorHandling("WSAStartup() error");

	// 1. IO Completion Port 객체 (CP 오브젝트)를 생성한다.
	hComPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
	GetSystemInfo(&sysInfo);
	// CP 오브젝트를 인수로 하는 IO 완료후 서비스를 제공하는 쓰레드를 실행한다. 
	for (int i = 0; i < sysInfo.dwNumberOfProcessors; ++i)
	{
		_beginthreadex(NULL, 0, (_beginthreadex_proc_type)EchoThreadMain, (LPVOID)hComPort, 0, NULL);
	}
	// 2. 기본 Overlapped IO 소켓을 생성하고 주소 할당 및 소켓을 연결 대기 상태에 놓는다.
	hServSock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
	memset(&servAdr, 0, sizeof(servAdr));
	servAdr.sin_family = AF_INET;
	servAdr.sin_addr.s_addr = htonl(INADDR_ANY);
	servAdr.sin_port = htons(PORT);

	bind(hServSock, (SOCKADDR*)&servAdr, sizeof(servAdr));
	listen(hServSock, 5);

	while (1)
	{
		SOCKET hClntSock;
		SOCKADDR_IN clntAdr;
		int addrLen = sizeof(clntAdr);

		// 3. 소켓에 클라이언트 요청이 들어오면 accept 함수는 클라이언트 핸들을 반환한다.
		hClntSock = accept(hServSock, (SOCKADDR*)&clntAdr, &addrLen);

		// 4. "소켓" 정보를 담는 구조체를 동적 할당하고 연결된 클라이언트 정보로 채운다.
		handleInfo = (LPPER_HANDLE_DATA)malloc(sizeof(PER_HANLDE_DATA));
		handleInfo->hClntSock = hClntSock;
		memcpy(&handleInfo->clntAdr, &clntAdr, addrLen);

		// 5. 클라이언트 소켓과 CP 오브젝트를 연결하고 CompletionKey 인자에 "소켓" 정보 구조체를 전달한다.
		CreateIoCompletionPort((HANDLE)hClntSock, hComPort, (DWORD)handleInfo, 0);

		// 6. "메세지 버퍼"를 담는 구조체를 동적 할당하고 WSARecv 함수를 호출하여 중첩 입력 동작을 수행한다.
		ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
		memset(&ioInfo->overlapped, 0, sizeof(OVERLAPPED));
		ioInfo->wsaBuf.len = BUF_SIZE;
		ioInfo->wsaBuf.buf = ioInfo->buffer;
		ioInfo->rwMode = READ;
		WSARecv(handleInfo->hClntSock, &(ioInfo->wsaBuf), 1, (LPDWORD)&recvBytes, (LPDWORD)&flags, &(ioInfo->overlapped), NULL);
	}
	return 0;
}

 

EchoThreadMain 함수

EchoThreadMain 함수에서 처리되는 로직은 다음과 같다.

1. CP 오브젝트에 등록되어 있는 소켓 핸들에 대한 IO 입출력 완료가 완료되면 블로킹 상태였던 GetQueueCompletionStatus가 함수를 반환한다. 

2. handleInfo와 ioInfo를 주소로 전달하여 main 함수에서 전달한 "메세지 버퍼"와 "소켓" 정보를 가져온다.

3. ioInfo에 기록해 두었던 IO 종류 (READ,WRITE)를 구분하여 클라이언트의 에코 메세지 전송 / 메세지 수신 / 연결 종료 등의 서비스를 제공한다.

DWORD __stdcall EchoThreadMain(LPVOID pComPort)
{
	HANDLE hComPort = (HANDLE)pComPort;
	SOCKET sock;
	DWORD bytesTrans;
	LPPER_HANDLE_DATA handleInfo;
	LPPER_IO_DATA ioInfo;
	DWORD flags = 0;

	while (1)
	{
		// 1. CP 오브젝트에 등록되어 있는 소켓 핸들에 대한 IO 입출력 완료가 완료되면 블로킹 상태였던 GetQueueCompletionStatus가 함수를 반환한다. 
		GetQueuedCompletionStatus(hComPort, &bytesTrans, (LPDWORD)&handleInfo,
			(LPOVERLAPPED*)&ioInfo, INFINITE);
		// 2. handleInfo와 ioInfo를 주소로 전달하여 main 함수에서 전달한 "메세지 버퍼"와 "소켓" 정보를 가져온다.
		sock = handleInfo->hClntSock;
		// 3. ioInfo에 기록해 두었던 IO 종류 (READ,WRITE)를 구분하여 클라이언트의 에코 메세지 전송 / 메세지 수신 / 연결 종료 등의 서비스를 제공한다.
		if (ioInfo->rwMode == READ)
		{
			puts("message received!");
			// 연결 종료
			if (bytesTrans == 0) // EOF 전송시
			{
				closesocket(sock);
				free(handleInfo); free(ioInfo);
				continue;
			}

			// 메세지 송신 : 중첩 출력과 입력을 수행
			memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
			ioInfo->wsaBuf.len = bytesTrans;
			ioInfo->rwMode = WRITE;
			WSASend(sock, &ioInfo->wsaBuf, 1, NULL, 0, &(ioInfo->overlapped), NULL);

			ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
			memset(&ioInfo->overlapped, 0, sizeof(OVERLAPPED));
			ioInfo->wsaBuf.len = BUF_SIZE;
			ioInfo->wsaBuf.buf = ioInfo->buffer;
			ioInfo->rwMode = READ;
			WSARecv(sock, &ioInfo->wsaBuf, 1, NULL, &flags, &ioInfo->overlapped, NULL);
		}

		else 
		{
			puts("message sent!");
			free(ioInfo);
		}
	}

	return 0;
}