AccepterThread 이해하기

프로젝트 : https://github.com/HoonInPark/ServerHyperion.git
본 포스트에 대한 내용은 feat/framesync 브랜치에 있다.

문제

현재 서버에서 Client Info 객체의 풀링을 구현하고 있는데,
접속이 들어오면 풀에서 디큐해서 unordered_set에 넣도록 수정하려 한다.
근데 예제로 받은 채팅 서버에서 AccepterThread의 의도를 잘 모르겠다.

현재는 Client Info를 관리하는 방식은 접속 여부와 상관 없이
vector<stClientInfo> m_ClientInfos에 담고 있다.

void AccepterThread()
{
	while (mIsAccepterRun)
	{
		auto curTimeSec = chrono::duration_cast<chrono::seconds>(chrono::steady_clock::now().time_since_epoch()).count();

		for (auto client : m_ClientInfos)
		{
			// if client elem is in use, continue.
			if (client->IsConnected())
			{
				continue;
			}

			if ((UINT64)curTimeSec < client->GetLatestClosedTimeSec())
			{
				continue;
			}

			auto diff = curTimeSec - client->GetLatestClosedTimeSec();
			if (diff <= RE_USE_SESSION_WAIT_TIMESEC)
			{
				continue;
			}

			client->PostAccept(mListenSocket, curTimeSec);
		}

		this_thread::sleep_for(chrono::milliseconds(32));
	}
}

해결

서버 시각 측정, 왜 하는 거지?

GetLatestClosedTimeSec()는 stClientInfo 내의 mLatestClosedTimeSec를 반환하는 함수.
mLatestClosedTimeSec는...

1. stClientInfo가 접속될 때

mLatestClosedTimeSec = UINT32_MAX;

와 같이 디폴트 값이 담기고,

2. 접속이 끊길 때

mLatestClosedTimeSec = chrono::duration_cast<chrono::seconds>(chrono::steady_clock::now().time_since_epoch()).count();

와 같이 값이 담긴다.
끊길 때 시각을 담는 멤버인 것.

이 값을 가지고...
현재 시각이 이 값 보다 작으면 continue.

if ((UINT64)curTimeSec < client->GetLatestClosedTimeSec())
{
	continue;
}

만약 서버에 접속한 시간이 RE_USE_SESSION_WAIT_TIMESEC 보다 작거나 같으면 또 continue.

auto diff = curTimeSec - client->GetLatestClosedTimeSec();
if (diff <= RE_USE_SESSION_WAIT_TIMESEC)
{
	continue;
}

즉, 서버 시간 측정은, Client Info가 접속이 끊겼을 때 시각과 비교하기 위함.
연결이 끊겼어도 끊긴 시각으로부터 일정시간이 지나야 Client Info를 재사용할 수 있도록 설계한 것임.
그래도 RE_USE_SESSION_WAIT_TIMESEC를 임의로 3초라고 정한 건 좀 못미덥다...

그리고 PostAccept(SOCKET, const UINT64) 호출

위 분기에 모두 해당되지 않은 경우, 즉 사용해도 되는 Client Info 객체인 경우,
Client Info에 있는 PostAccept(SOCKET, const UINT64)가 호출된다.

bool PostAccept(SOCKET listenSock_, const UINT64 curTimeSec_)
{
	printf_s("PostAccept. client Index: %d\n", GetIndex());

	mLatestClosedTimeSec = UINT32_MAX;

	mSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_IP,
		NULL, 0, WSA_FLAG_OVERLAPPED);
	if (INVALID_SOCKET == mSocket)
	{
		printf_s("client Socket WSASocket Error : %d\n", GetLastError());
		return false;
	}

	ZeroMemory(&mAcceptContext, sizeof(stOverlappedEx));

	DWORD bytes = 0;
	DWORD flags = 0;
	mAcceptContext.m_wsaBuf.len = 0;
	mAcceptContext.m_wsaBuf.buf = nullptr;
	mAcceptContext.m_eOperation = IOOperation::IO_ACCEPT;
	mAcceptContext.SessionIndex = mIndex;

	if (FALSE == AcceptEx(listenSock_, mSocket, mAcceptBuf, 0,
		sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16, &bytes, (LPWSAOVERLAPPED) & (mAcceptContext)))
	{
		if (WSAGetLastError() != WSA_IO_PENDING)
		{
			printf_s("AcceptEx Error : %d\n", GetLastError());
			return false;
		}
	}

	return true;
}

여기선 AcceptEx(...) 함수로 mAcceptContext가 연결을 받도록 예약을 걸어둔다.
여기로 연결이 완료되면 커널은 내부적으로 이벤트 큐에 IO_ACCEPT에 해당하는 이벤트를 엔큐.

void WokerThread()
{
	//CompletionKey를 받을 포인터 변수
	stClientInfo* pClientInfo = nullptr;
	//함수 호출 성공 여부
	BOOL bSuccess = TRUE;
	//Overlapped I/O작업에서 전송된 데이터 크기
	DWORD dwIoSize = 0;
	//I/O 작업을 위해 요청한 Overlapped 구조체를 받을 포인터
	LPOVERLAPPED lpOverlapped = NULL;

	while (mIsWorkerRun)
	{
		bSuccess = GetQueuedCompletionStatus(
			mIOCPHandle,
			&dwIoSize,					// 실제로 전송된 바이트
			(PULONG_PTR)&pClientInfo,		// CompletionKey
			&lpOverlapped,				// Overlapped IO 객체
			INFINITE);					// 대기할 시간
		
        // ...
        
		switch (pOverlappedEx->m_eOperation)
		{
		case IOOperation::IO_ACCEPT:
		{
			pClientInfo = GetClientInfo(pOverlappedEx->SessionIndex);
			if (pClientInfo->AcceptCompletion())
			{
				//클라이언트 갯수 증가
				++mClientCnt;
				OnConnect(pClientInfo->GetIndex());
			}
			else
			{
				CloseSocket(pClientInfo, true);
			}

			break;
		}
        
        // ...
        
	}
}

위 IO Thread의 GetQueuedCompletionStatus(...)함수는 커널이 관리하는 이벤트 큐에서 디큐한다.
큐가 비어 있으면 무한 대기(INFINITE).