[C++] Completion Port Model

대망의 IOCP.

이번에도 최대한 기억을 짜내서 메모해 보자.

 

 

1. Completion Model의 개념

Overlapped 모델을 생각해 보자.

이 모델 APC큐가 스레드마다 있었다.

따라서 어찌 보면 스레드가 낭비되는 느낌이 있었는데, IOCP는 그렇지 않다.

APC큐가 스레드마다 있지 않고 딱 1개가 있다.

중앙에서 관리하는 APC큐라는 느낌으로 보면 딱 맞을 것이다.

 

Alertable Wait 상태로 들어가기 위해 GetQueuedCompletionStatus()라는 함수를 사용해야 한다.

여기까지만 보면 Overlapped 모델과 흐름상의 큰 차이는 없지만,

결과적으로 IOCP는 스레드와의 궁합이 매우 좋기 때문에 매우 성능이 좋은 모델이라 할 수 있다.

 

 

1-1. 사용할 함수

CreateIoCompletionPort()와 앞서 얘기한 GetQueuedCompletionStatus()가 사용될 것이다.

 

각 함수의 역할은 아래와 같다.

• CreateIoCompletionPort()
  - Completion Port Handle을 만드는 것과 소켓을 CP에 등록하는 일을 하나의 함수가 수행한다.
• GetQueuedCompletionStatus()
  - 실질적으로 결과 처리를 감지한다.

 

1-2. 사전 작업

여기에 살짝의 변화가 필요하다.먼저 Session 구조체에서 WSAOVERLAPPED를 분리해 새로운 OverlappedEx 구조체를 만든다.그리고 이번엔 에코서버가 아니기에 받기만 할 것이라 send와 관련한 녀석들도 빼준다.

 

struct Session
{
	// WSAOVERLAPPED overlapped = {}; <<- 이게 빠졌다!
	SOCKET socket = INVALID_SOCKET;
	char recvBuffer[BUFSIZE] = {};
	int32 recvBytes = 0;	
};

최종적으로 Session 구조체는 위와 같은 형태가 될 것이다.

 

새로 만들 OverlappedEx 구조체엔 type이라는 새로운 녀석도 넣어줄 것이다.

// Session에 있던 overlapped를 가져옴
struct OverlappedEx
{
	WSAOVERLAPPED overlapped = {};
	int32 type = 0; // read, write, accept, connect ...
};

type은 얘가 어떤 작업을 수행할 것인지 구분짓기 위해 필요한 변수다.

이 구분에 맞춰 enum을 하나 만들어 준다.

 

enum IO_TYPE
{
	READ,
	WRITE,
	ACCEPT,
	CONNECT,
};

이정도면 충분할 것이다.

 

 

2. IOCP 모델 작성

역시 이번에도 기존에 소켓을 만드는 부분은 생략한다.

 

먼저 세션을 갖고 있을 매니저를 벡터로 하나 만들고,

IOCP 핸들을 생성한 후, Accept를 수행하고 소켓을 CP에 등록하는 것까지 해보자.

 

// 모든 세션을 들고 있을 예정
vector<Session*> sessionManager;

// CP 생성, 다 기본 상태로 밀어 둠
HANDLE iocpHandle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

// Main Thread = Accept 담당
// 원래 Accept도 CP로 처리하지만 여기선 메인 스레드에서 처리
// IO는 별도의 스레드에서 처리
while (true)
{
	SOCKADDR_IN clientAddr;
	int32 addrLen = sizeof(clientAddr);

	SOCKET clientSocket = ::accept(listenSocket, (SOCKADDR*)&clientAddr, &addrLen);
	if (clientSocket == INVALID_SOCKET)
		return 0;

	// 세션을 얼마나 들고 있게 될지 모르기 때문에
	// 동적 할당을 통해 스택이 아니라 힙에 생성
	Session* session = new<Session>();
	session->socket = clientSocket;
	sessionManager.push_back(session);

	cout << "Client Connected !" << endl;

	// 소켓을 CP에 등록
	// Completion Key
	// - GetQueuedCompletionStatus를 통해 일감을 빼올 때,
	// - 어떤 것인지 알기 위한(식별) 고유의 키값
	// NumberOfConcurrentThreads
	// - 0으로 밀면 알아서 최대 개수의 코어를 사용
	// - 스레드를 알아서 생성해 주는 것은 아님
	::CreateIoCompletionPort((HANDLE)clientSocket, iocpHandle, /*Key*/(ULONG_PTR)session, 0);
	
    // ...
}

 

CreateIoCompletionPort()를 활용해 핸들을 생성하는 것을 볼 수 있다.

기본적으로 핸들을 생성하기 위해 NULL과 0 등으로 밀어줬다.

 

세션을 생성하고 매니저에 넣어줬는데, 이번에 세션은 동적 할당을 통해 힙 영역에 올려줬다.

세션이 얼마나 생길지 어떻게 알겠는가?

 

그리고 다시 CreateIoCompletionPort()를 사용해 소켓을 핸들에 등록했다.

확실히 하나의 함수가 2가지 일을 수행한다는 것이 그렇게 잘 와닿진 않는다.

 

일단 첫 인자로 HANDLE로 캐스팅한 소켓을 넘겨준다. 이제 이 소켓을 관찰할 것이다.

다음으로 iocpHandle을 넘겨줬는데, 이는 ExistingCompletionPort로 아까 CP 핸들을 생성해 줬으니,

존재하는 CP로 취급된다. 따라서 넘겨준다.

다음으론 ULONG_PTR 형식의 Key인데, 여기엔 정말 아무 값이나 넘겨줄 수 있다.

하지만 이 Key로 나중에 식별을 해야 하기 때문에 정말 아무 값을 넣는 것보단,

sessions의 주소를 캐스팅하여 넘겨주어 식별에 용이하게끔 한다.

 

마지막은 얼마나 많은 스레드로 일을 할지에 대한 값인데,

0으로 밀면 알아서 가용한 최대 개수의 코어를 끌어와 일을 한다.

하지만 알아서 스레드를 만들어서 일을 시키는 것은 아니다.

 

이제 나머지 부분을 작성해 보자.

WSABUF wsaBuf;
wsaBuf.buf = session->recvBuffer;
wsaBuf.len = BUFSIZE;

OverlappedEx* overlappedEx = new OverlappedEx();
overlappedEx->type = IO_TYPE::READ;

// 키값으로 이미 세션을 넘겨줬고
// overlapped 구조체로 데이터를 한번 더 넘겨줄 수 있다.
DWORD recvLen = 0;
DWORD flags = 0;
::WSARecv(clientSocket, &wsaBuf, 1, &recvLen, &flags, &overlappedEx->overlapped, NULL);

넘겨줄 버퍼 구조체를 선언하고, 아까 작성한 OverlappedEx 구조체도 선언해 준다.

에코 서버가 아니라 읽기만 할 것이므로 type은 READ를 줬다.

그다음 recv를 걸어 GetQueuedCompletionStatus()를 통해 감지 결과를 받을 수 있게 만든다.

별도의 콜백 함수가 없으므로 마지막 인자는 NULL로 민다.

 

이러면 메인 스레드에서 할 일에 대한 지시는 끝났다.

 

 

2-1. Worker 작성

메인 스레드만으론 지속적인 통신이 불가능하다.

워커 스레드에서 실제로 처리할 부분을 담당할 함수를 만들어 주자.

 

void WorkerThreadMain(HANDLE iocpHandle)
{
	while (true)
	{
		DWORD bytesTransferred = 0;
		Session* session = nullptr;
		OverlappedEx* overlappedEx = nullptr;

		BOOL ret = ::GetQueuedCompletionStatus(iocpHandle, &bytesTransferred,
			(ULONG_PTR*)&session, (LPOVERLAPPED*)&overlappedEx, INFINITE);

		if (ret == FALSE || bytesTransferred == 0)
		{
			// TODO : 연결 끊김
			continue;
		}
        
        // ...
	}
}

GetQueuedCompletionStatus()에 넘겨줄 변수들을 몇 개 만들어 주자.

포인터 형식으로 결과값들을 받아올 수 있게 한다.

 

리턴값이 FALSE거나 받은 바이트가 0이면 문제 상황으로 간주하고 그 후속 처리를 해 주게 된다.

여기선 별도 처리 없이 continue로 진행하게 했다.

 

// 실제론 분기문을 통해 구분해서 처리하게 됨
// 여기선 READ만 할 것이므로 이것만.
ASSERT_CRASH(overlappedEx->type == IO_TYPE::READ);

cout << "Recv Data Len IOCP = " << bytesTransferred << endl;
cout << "Recv Data IOCP = " << session->recvBuffer << endl;

WSABUF wsaBuf;
wsaBuf.buf = session->recvBuffer;
wsaBuf.len = BUFSIZE;

DWORD recvLen = 0;
DWORD flags = 0

// 여기서 직접 Recv를 호출해 계속 반복할 수 있도록 한다
// 낚아올리고 다시 낚싯대를 드리우는 것과 같다
::WSARecv(session->socket, &wsaBuf, 1, &recvLen, &flags, &overlappedEx->overlapped, NULL);

원래 READ나 WRITE 등을 구분해서 if else나 스위치 문을 통해 별도의 처리 루틴을 다 만들어 줘야 하지만...

이번엔 받기만 할 것이기 때문에 READ에 대해서만 처리한다.

그래도 일단 예외처리는 해준다.

 

이제 받은 데이터의 길이와 그 내용을 출력한다.

 

이제 다시 버퍼와 기타 변수들을 만들어서 WSARecv()를 실행시킨다.

여기서 WSARecv를 실행시켜 주어야 모델이 멈추지 않고 계속 돌아갈 수 있다.

IO 함수를 실행해 결과 처리를 감지하게끔 해 주어야만 한다!

 

여기서 overlapped를 재활용했는데, 똑같이 READ에 사용할 것이기 때문에 문제가 없다.

만약 WRITE 등의 다른 IO 함수를 사용한다면 별도의 overlapped 변수를 할당해 넘겨주어야만 한다.

재사용할 것이 아니라면 메모리 해제를 시켜주면 될 것이다.

 

2-2. 스레드 생성

자 다시 메인 스레드로 돌아와서 워커를 돌려줄 스레드를 만들어야 한다.

여기선 이전에 공부할 때 만들었던 별도의 스레드 매니저를 활용해 보자.

 

// WorkerThreads
for (int32 i = 0; i < 5; i++)
	GThreadManager->Launch([=]() { WorkerThreadMain(iocpHandle); });

람다를 활용해 iocpHandle을 복사해서 워커 스레드를 생성해 가는 형태로 스레드를 생성한다.

그리고 스레드가 일을 다 하기 전에 프로그램이 종료되는 것을 막기 위해 반복문 밖에 Join 하는 코드도 넣어줬다.

 

대충 동작에 필요한 부분은 작성한 것 같다.

실행해 보자.

 

3. 실행

잘 보내고 워커 스레드에서 잘 받아준다.

 

 

4. 생각해 볼 점

사실 여기엔 큰 문제가 있다.

만약 클라이언트 세션이 예기치 못하게 종료된다면?

이미 넘겨준 세션의 주소값이 오염 돼버린다면?

이러면 서비스에 정말 큰 문제가 생길 것이다.

 

그렇다면 세션을 어떻게 관리해 주어야 할까?

세션이 종료되면 그 부분은 필연적으로 쓰레기 값이 되어 버린다.

그러면 스마트 포인터처럼 카운트로 판단해,

해당 세션의 역할이 다 하기 전까지 지울 수 없게끔 하면 되지 않을까 라는 아이디어가 떠오른다.

 

일단 아래의 코드를 넣어서 갑자기 세션이 죽는 상황을 테스트해 보자.

Session* s = sessionManager.back();
sessionManager.pop_back();
delete(s);

 

4-1. 세션이 죽었을 때 결과

 

갑자기 세션이 죽어버려서 오염된 메모리를 참조해 버렸기 때문에 서버가 더 이상 진행되지 않는다.

그리고 분명 오염된 메모리를 참조했는데, 크래시 되지 않는 부분도 확인할 수 있다.

정말 무서운 부분 중 하나인데, 이렇게 오염된 메모리를 확인하지 못하다가 엉뚱한 곳에서 문제가 터지면,

문제의 원인을 잡는 것부터가 쉽지가 않다.

 

일단 레퍼런스 카운트가 되게끔 동적할당 할 수 있게 하자.

여기서도 저번에 공부하며 이런 때를 대비해 작성했던 라이브러리를 사용해 보자.

 

//Session* session = new Session();
Session* session = xnew<Session>();

기존 new로 할당했던 세션을 xnew()로 할당한다.

 

//delete(s)
xdelete(s);

delete도 xdelete로 변경.

 

그리고 중요한 한 가지.

Stomp Allocator를 활성화해 박살 난 메모리 상황에 대해 크래시를 뱉을 수 있도록 한다.

 

 

다시 실행해 보자.

 

오염된 메모리를 참조한 것에 대해 크래시를 뱉어주게 된다.

 

 

5. 마무리

어렵다. 확실히 이 흐름을 이해하기 까진 역시 시간이 걸릴 듯하다.

내가 틀렸다면 정확히 뭘 틀렸는지도 감이 오지 않기 때문에

틀린 점이 있다면 공부해 나가면서 수정토록 해야만 하겠다.

 

이제 IOCP를 활용한 네트워크 라이브러리를 작성하는 것에 대한 공부를 시작하게 된다.

만만하진 않겠지만, 새로이 뭔가 제대로 된 걸 배운다는 것은 항상 기대되는 일인 것 같다.