[WINAPI] 비동기 소켓 재활용 예제

소켓 재활용은 이미 연결이 끊어진 소켓을 재사용하는 기술입니다. 소켓을 재활용하면 새로운 소켓을 생성하지 않고 기존 소켓을 활용하여 새로운 연결을 맺을 수 있습니다. 즉 소켓 생성 및 소켓 핸들 관리에 필요한 시스템 리소스를 절약할 수 있습니다. 소켓을 재활용하기 위해서는 소켓 연결이 끊어진 상태여야 합니다.

 

서버 관점에서 클라이언트의 접속을 비동기로 처리하는 과정은 다음과 같습니다.

1. 리슨 소켓 생성 및 리슨 포트 바인드
2. 소켓 미리 생성 후 대기 소켓 풀에 등록
3. 클라이언트 접속 종료 시 사용한 소켓 재사용처리 후 대기 소켓 풀에 등록

 

예제 코드는 IOCP 기반입니다. 이해가 되지 않으면 다음의 IOCP 프로토타입 포스팅을 참고하세요.

2023.12.13 - [프로그래밍/C | C++] - IOCP 기반 다중 접속 및 데이터 처리-1

IOCP 기반 다중 접속 및 데이터 처리-1

2023.12.14 - [프로그래밍/C | C++] - IOCP 기반 다중 접속 및 데이터 처리-2

IOCP 기반 다중 접속 및 데이터 처리-2

 

1. LISTEN SOCKET 생성하기

만약 443번 포트로 서비스를 하려면 443번의 포트로의 접속 요청을 현재 프로세스가 소유한 소켓 핸들과 연관 관계를 맺어 주어야 합니다. 이를 소켓 바인딩이라고 합니다. 빈 소켓을 생성 후 포트번호와 함께 바인드함수를 호출하면 443번 포트는 현재 프로세스의 소켓 핸들과 연결되어 컨트롤 가능해집니다. 그다음 소켓을 LISTEN 상태로 설정하여 클라이언트의 연결 요청을 받을 수 있는 상태로 변경하면 리슨 소켓 생성 단계가 완료됩니다.

// TCP/IP 속성의 소켓을 생성
SOCKET hsoListen = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

// 소켓 생성 확인
if (hsoListen == INVALID_SOCKET)
{
	DPRINTF("[ERR] socket failed, code : (%d)\n", WSAGetLastError());
	return INVALID_SOCKET;
}

// shPortNo: 포트번호
// 위에서 생성한 소켓을 포트번호와 bind
SOCKADDR_IN	sa;
memset(&sa, 0, sizeof(SOCKADDR_IN));
sa.sin_family = AF_INET;
sa.sin_port = htons(shPortNo);
sa.sin_addr.s_addr = htonl(INADDR_ANY);
LONG lSockRet = bind(hsoListen, (PSOCKADDR)&sa, sizeof(SOCKADDR_IN));
if (lSockRet == SOCKET_ERROR)
{
	DPRINTF("[ERR] bind failed, code : (%d)\n", WSAGetLastError());
	closesocket(hsoListen);
	return INVALID_SOCKET;
}
DPRINTF("INFO] listen() setup with nBacklog-SOMAXCONN (%x)-(%x)\n", nBacklog, SOMAXCONN);

// 포트를 리슨상태로 변경
lSockRet = listen(hsoListen, nBacklog);
if (lSockRet == SOCKET_ERROR)
{
	DPRINTF("[ERR] listen failed, code : (%d)\n", WSAGetLastError());
	closesocket(hsoListen);
	return INVALID_SOCKET;
}

// shPortNo 포트로 접속 대기 상태가 됨

 

2. IOCP에 LISTEN SOCKET 등록하기

생성된 LISTEN SOCKET에 클라이언트 접속 이벤트가 발생하면 IOCP 콜백 함수가 호출되도록 LISTEN SOCKET으로 IOCP 핸들을 생성합니다. 다음의 코드는 IOCP 핸들을 생성하는 동시에 파라미터로 전달한 hsoListen 소켓 핸들을 IOCP에 등록하고 IOCP 콜백 호출 시 IOKEY_LISTEN 값을 파라미터로 전달합니다.

#define IOKEY_LISTEN	1

HANDLE hIocp = INVALID_HANDLE_VALUE;

hIocp = CreateIoCompletionPort((HANDLE)hsoListen, NULL, IOKEY_LISTEN, (stSysInfo.dwNumberOfProcessors / 2));
if (NULL == hIocp) {
	DPRINTF("[ERR] CreateIoCompletionPort failed, code : (%d)\n", GetLastError());
	return 0;
}

 

3. IOCP LISTEN SOCKET에 ACCEPT 소켓 미리 생성하여 할당하기

Accept() 함수는 동기함수이기 때문에 호출시 클라이언트가 접속할 때까지 호출위치에서 블로킹상태에 빠집니다.

클라이언트의 접속 요청을 비동기적으로 처리하기 위해서는 소켓을 미리 생성 후 IOCP에 등록된 LISTEN SOCKET과 BIND 해두면 접속 이벤트 발생 시 IOCP 콜백함수가 호출되게 됩니다.

// AcceptEx 함수 포인터 획득
LPFN_ACCEPTEX pfnAcceptEx = (LPFN_ACCEPTEX)
	GetSockExtAPI(hsoListen, WSAID_ACCEPTEX);
   
int nPooledCnt = 0;
int nIncCnt = 10;	// 미리 생성할 소켓 수 10개로 설정
for (; nPooledCnt < nIncCnt; nPooledCnt++)
{
	// 빈 소켓을 미리 생성한다.
	SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (sock == INVALID_SOCKET)
		break;
        
    // OVERLAPPED 구조체를 상속한 소켓 노드 구조체
    // typedef struct _ril_session_node : OVERLAPPED
    // 구조체 내부에 SOCKET _sock; 을 필드로 갖고 있음
	PTST_RIL_SESSION_NODE prsn = new TST_RIL_SESSION_NODE;
	prsn->_sock = sock;
    
    // AcceptEx 호출. 리슨소켓에 빈 소켓을 바인딩함과 동시에 OVERLAPPED 구조체를 넘겨서
    // listen socket이 등록된 IOCP 핸들과 연결 한다.
    // 즉, listen socket에 접속 이벤트가 발생하면 미리 생성한 빈 소켓에 Accept가 되면서
    // IOCP 콜백함수가 호출되며 이 때의 파라미터는 CreateIoCompletionPort 호출시에 넘긴
    // IOKEY_LISTEN 값이 되므로 이를 통해 새로운 접속이벤트를 처리할 수 있다.
	BOOL bIsOK = pfnAcceptEx
	(
		hsoListen, sock, prsn->_buff, ACCEPTEX_INITIAL_DWRECEIVEDATALEN,
		sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16,
		NULL, (LPOVERLAPPED)prsn
	);
    
	if (bIsOK == FALSE)
	{
		if (WSAGetLastError() != WSA_IO_PENDING)
		{
			DPRINTF("[ERR] AcceptEx failed, code : (%d)\n", WSAGetLastError());
			closesocket(prsn->_sock);
			delete prsn;
			break;
		}
	}
    
    // 미리 생성해둔 소켓은 별도 소켓 풀에 등록해 둠
	AcquireSRWLockExclusive(&(_m_stSessionObj.srwlock));
	_m_stSessionObj.mapAcceptExPool.SetAt(prsn, prsn->_sock);
	ReleaseSRWLockExclusive(&(_m_stSessionObj.srwlock));
}

 

4. IOCP 콜백 함수에서 클라이언트 Connect, Disconnect 감지

ThdRilIocp() 는 IOCP 스레드풀의 Worker Thread입니다. 이벤트 발생 시 호출되는 Callback Thread입니다.

Connect  감지는 IOKEY_LISTEN 값을 파라미터로 받았을 때 입니다. Accept 완료 상태이므로 Accept 소켓을 IOCP에 등록합니다. 접속한 클라이언트가 패킷 데이터를 송신하면 역시 IOCP 콜백스레드가 호출되며 else {} 구문으로 빠지게 됩니다. else {} 구문 진입했는데 전송받은 데이터가 0바이트이면 접속 종료로 처리하고 그게 아니면 데이터를 처리하면 됩니다.

 

UINT CRilServer::ThdRilIocp(LPVOID pParam)
{
...생략...

	PTST_RIL_SESSION_NODE prsn = NULL;

	int nErrCode;

	while (TRUE) {
		try
		{
			BOOL bIsOK = GetQueuedCompletionStatus(hIocp, &dwTrBytes, &upDevKey, (LPOVERLAPPED*)&prsn, INFINITE);

			if (FALSE == bIsOK)
			{
				...생략...
			}

			// 신규 클라이언트 Connect 이벤트 감지 부분
			if (IOKEY_LISTEN == upDevKey)
			{
            	// 신규 접속이므로 미리 생성된 소켓을 IOCP에 등록해주고 이벤트 키는 IOKEY_CHILD로 등록
				CreateIoCompletionPort((HANDLE)prsn->_sock, hIocp, IOKEY_CHILD, 0);
				DPRINTF("[INFO] ==> New client, connected...\n");
				pRilObj->_m_pThdRilServerMgr->PostThreadMessageA(MSG_ID_CRILSEREVR_SOCK_CONNECTED, 0, (LPARAM)prsn);
			}
			else
			{
            	// 클라이언트 Disconnect 이벤트 감지 부분
				if (0 == dwTrBytes) {
					throw (INT)ERROR_SUCCESS;	/* Socket disconnected from client */
				}
                
               ...생략...
               
             }
....생략
}

 

catch 문에서 처리하는 STATUS 값은 다음과 같습니다.

#ifndef NTSTATUS_SOCK_LOCAL_DISCONNECT
#	define NTSTATUS_SOCK_LOCAL_DISCONNECT	((LONG)0xC000013BL)	//ERROR_NETNAME_DELETED
#endif
#ifndef NTSTATUS_SOCK_REMOTE_DISCONNECT
#	define NTSTATUS_SOCK_REMOTE_DISCONNECT	((LONG)0xC000013CL)	//ERROR_NETNAME_DELETED
#endif
#ifndef NTSTATUS_SOCK_CONNECTION_RESET
#	define NTSTATUS_SOCK_CONNECTION_RESET	((LONG)0xC000020DL)	//ERROR_NETNAME_DELETED
#endif
#ifndef NTSTATUS_SOCK_CANCELLED
#	define NTSTATUS_SOCK_CANCELLED			((LONG)0xC0000120L)	//ERROR_OPERATION_ABORTED
#endif

catch (int ex)
{
	if ((ex == NTSTATUS_SOCK_LOCAL_DISCONNECT) || (ex == NTSTATUS_SOCK_CANCELLED))
	{
		DPRINTF("[DBG]  ==> Child socket closed.\n");
		continue;
	}
	if ((ex == ERROR_SUCCESS) || (ex == NTSTATUS_SOCK_REMOTE_DISCONNECT)) {
		DPRINTF("[DBG]  ==>  ==> Client  disconnected...\n");
	}
	else if (ex == NTSTATUS_SOCK_CONNECTION_RESET) {
		DPRINTF("[DBG]  ==> Pending Client disconnected...\n");
	}
	else {
		DPRINTF("[DBG] ==> Client has error(%d)...\n", ex);
	}
	pRilObj->_m_pThdRilServerMgr->PostThreadMessageA(MSG_ID_CRILSERVER_SOCK_DISCONNECTED, ex, (LPARAM)prsn);
}

 

5. 접속, 종료 세션 관리

저는 접속 세션 객체를 리스트로 관리했는데 세션을 관리하는 자료구조는 프로그램 특성에 따라 자유롭게 개발하면 됩니다. IOCP 스레드에서 접속, 종료에 따라 이벤트 메시지를 전송하면 아래와 같이 세션에 등록하거나 세션에서 해제합니다.

PTST_RIL_SESSION_NODE prsn = (PTST_RIL_SESSION_NODE)msg.lParam;
if (MSG_ID_CRILSEREVR_SOCK_CONNECTED == msg.message) {
	pthis->m_RilSession.vRegisterSocketToSessionObj(prsn);
}
else if (MSG_ID_CRILSERVER_SOCK_DISCONNECTED == msg.message) {
	pthis->m_RilSession.vUnregisterSocketFromSessionObj(hsoListen, prsn);
}

 

6. 소켓 재사용 처리

종료세션 함수 vUnregisterSocketFromSessionObj()의 코드 중 소켓 재사용 처리하는 부분입니다. 

close 함수 대신 disconnectex를 통해 접속만 종료 후 다시 acceptex를 호출하고 있습니다.

재사용이기 때문에 상위 응용 스레드에서는 소켓 핸들값만 가지고 클라이언트의 변경을 감지하기 어렵습니다.

따라서 세션을 관리하는 자료구조에서는 클라이언트 접속시 클라이언트 고유 번호를 부여하고

접속 종료 이벤트 발생시 해당 클라이언트가 처리 중인 송수신 패킷을 다른 클라이언트에 이관하거나 상위 응용 스레드에 Abort 이벤트를 전달하고 클라이언트 고유번호를 갱신해야 합니다. 

예를 들어, 클라이언트 A가 서버로 데이터를 요청하여 서버가 데이터를 준비하고 있는 동안 클라이언트 A가 접속 종료를 해버리고 A가 사용하던 소켓 핸들을 재사용하여 클라이언트 B가 접속했는데 서버가 A에 응답하려던 패킷을 클라이언트 B가 받지 않도록 처리를 해야 합니다.

// 소켓을 close 하지 않고 disconnect 합니다.
LPFN_DISCONNECTEX pfnDisconnectEx = (LPFN_DISCONNECTEX)GetSockExtAPI(prsn->_sock, WSAID_DISCONNECTEX);
pfnDisconnectEx(prsn->_sock, NULL, TF_REUSE_SOCKET, 0);
prsn->pNext = nullptr;
prsn->pPrev = nullptr;
ZeroMemory(prsn->_buff, sizeof(TST_RIL_SESSION_NODE::_buff));
prsn->dwUsedLen = 0;
prsn->pStreamHead = nullptr;	/* should be called free function or pass pointer to worker thread before this line */
prsn->pStreamTail = nullptr;
		
ZeroMemory(&(prsn->saLocal), sizeof(SOCKADDR_IN));
ZeroMemory(&(prsn->saRemote), sizeof(SOCKADDR_IN));
ZeroMemory(&(prsn->sysTmLoggedIn), sizeof(SYSTEMTIME));
prsn->ullCntTx = 0;
prsn->ullCntRx = 0;
prsn->uRtt = 0;
ZeroMemory(prsn->tchNoyeNetId, sizeof(TST_RIL_SESSION_NODE::tchNoyeNetId));
prsn->uTxPriority = 0;
prsn->enApiCat = TEN_STOCK_API_CAT::UNKNOWN;
prsn->u8TradeSupportMask = 0;

// 다시 비동기 accept 호출하여 리슨 소켓에 등록해 두고 접속 이벤트 발생시 iocp 콜백 호출 되도록 합니다.
LPFN_ACCEPTEX pfnAcceptEx = (LPFN_ACCEPTEX)GetSockExtAPI(hsoListen, WSAID_ACCEPTEX);
boRet = pfnAcceptEx(hsoListen, prsn->_sock, prsn->_buff, ACCEPTEX_INITIAL_DWRECEIVEDATALEN, sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16,	NULL, (LPOVERLAPPED)prsn);
if (FALSE == boRet) {
	int nErrCode = WSAGetLastError();
	if (nErrCode != WSA_IO_PENDING)
	{
		DPRINTF("[ERR] AcceptEx failed, code : (%d)\n", GetLastError());
		closesocket(prsn->_sock);
		delete prsn;
		return;
	}
}

_m_stSessionObj.mapAcceptExPool.SetAt(prsn, prsn->_sock);
DPRINTF("DBG] Released socket. AcceptExPool(%Iu) LoggedinSessions(%Iu) ConnectedOnly(%Iu)\n", _m_stSessionObj.mapAcceptExPool.GetCount(), _m_stSessionObj.mapTchNoyeNetId.GetCount(), _m_stSessionObj.mapConnectedOnlyPool.GetCount());
}

 

7. 대기 소켓 풀 증가

예제 코드에서는 소켓 10개를 미리 생성하여 IOCP에 등록해두었습니다. 10개의 클라이언트까지는 IOCP 콜백이 호출되어 처리되지만 미리 생성한 소켓을 다 소진했을 때 추가적인 클라이언트 접속 요청을 받으면 미리 생성해 둔 소켓이 없기 때문에 LISTEN SOCKET에 이벤트가 발생하게 됩니다. IOCP 스레드가 아닌 관리 스레드에 LISTEN SOCKET에 대한 ACCEPT 이벤트를 대기하고 있다가 ACCEPT 이벤트 발생 시 소켓을 미리 생성하면 됩니다.

예제 코드에서의 핵심은 WSAEventSelect에서 FD_ACCEPT 플래그를 LISTEN SOCKET에 등록한 이벤트 hWSAEvent 핸들을 통해 ACCEPT를 감지하고 WSAEnumNetworkEvents에서 FD_ACCEPT 여부를 필터링하여 소켓 생성함수를 다시 호출하는 부분입니다. WSAEnumNetworkEvents 호출시 hWSAEvent 핸들에서 FD_ACCEPT 이벤트는 RESET 되어 다시 새로운 ACCEPT 이벤트를 감지할 수 있는 상태가 됩니다.

	WSAEVENT hWSAEvent = WSACreateEvent();
	WSAEventSelect(hsoListen, hWSAEvent, FD_ACCEPT);
	constexpr DWORD dwNCOUNT= 1;
	DWORD dwWaitRet;
	MSG msg;
	while (TRUE) {
		dwWaitRet = MsgWaitForMultipleObjectsEx(1, &hWSAEvent, RIL_SERVER_SOCKET_POOL_REFRESH_INTERVAL, QS_POSTMESSAGE, MWMO_INPUTAVAILABLE);

		if ((WAIT_OBJECT_0 + 0) == dwWaitRet) {	/* FD_ACCEPT event at hWSAEvent by hsoListen */
			WSANETWORKEVENTS ne;
			WSAEnumNetworkEvents(hsoListen, hWSAEvent, &ne);
			if (ne.lNetworkEvents & FD_ACCEPT) {

				int iRet;
				iRet = pthis->m_RilSession.IncreaseAcceptSockets(hsoListen);
				DPRINTF("hsoListen event signaled, iRet(%d) PoolCnt(%Iu), SessionCnt(%Iu)\n", iRet, pthis->m_RilSession.iGetCount_AcceptExPOOL(), pthis->m_RilSession.iGetCount_ConnectedSessions());
			}

 

여기까지 IOCP 기반의 소켓 접속, 종료, 재사용을 알아보았습니다.