얌생

구현한 채팅서버에서 패킷을 처리할때 패킷을 받으면서 처리했는데,

패킷을 처리할때 job을 생성하고, 해당 job을 UpdateThread에 넘겨서 그곳에서 처리하는 방식으로 변경했다.

iocp 환경상 많은 메세지를 처리해야하는데, packet을 바로 처리하기보다 updateThread에 넘겨서 처리하는 방식이 나을것이라고 판단했다.

✅ 수정 전

void CoreNetwork::RecvComplete(Session* recvCompleteSesion, const DWORD& transferred)
{
	int loopCount = 0;
	const int MAX_PACKET_LOOP = 64;
	recvCompleteSesion->recvRingBuffer.MoveRear(transferred);

	Packet::EncodeHeader encodeHeader;
	Packet* packet = Packet::Alloc();

	while (loopCount++ < MAX_PACKET_LOOP)
	{
		packet->Clear();

		// 최소한 헤더 크기만큼은 데이터가 왔는지 확인한다.
		if (recvCompleteSesion->recvRingBuffer.GetUseSize() < sizeof(Packet::EncodeHeader))
		{
			break;
		}
		
		// 헤더를 뽑아본다.
		recvCompleteSesion->recvRingBuffer.Peek((char*)&encodeHeader, sizeof(Packet::EncodeHeader));
		if (recvCompleteSesion->recvRingBuffer.GetUseSize() < encodeHeader.packetLen + sizeof(Packet::EncodeHeader))
		{
			break;
		}		

		// 헤더 크기만큼 front를 움직이기
		recvCompleteSesion->recvRingBuffer.MoveFront(sizeof(Packet::EncodeHeader));
		// 패킷 길이만큼 뽑아서 packet에 넣기
		recvCompleteSesion->recvRingBuffer.Dequeue(packet->GetRearBufferPtr(), encodeHeader.packetLen);
		// 헤더 설정
		packet->SetHeader((char*)&encodeHeader, sizeof(Packet::EncodeHeader));
		// 패킷 길이 만큼 rear 움직이기
		packet->MoveRearPosition(encodeHeader.packetLen);

		// 디코딩에 실패하면 연결을 끊음
		if (!packet->Decode())
		{
			Disconnect(recvCompleteSesion->sessionId);
			break;
		}	

		// 패킷 처리
		OnRecv(recvCompleteSesion->sessionId, packet);
	}
	
	packet->Free();

	RecvPost(recvCompleteSesion);
}

위 코드를 확인하면 알수 있듯이 packet을 받고 조립한 후 OnRecv()로 전달한다.

void ChattingServer::OnRecv(__int64 sessionId, Packet* packet)
{
	short protocol;
	*packet >> protocol;

	(this->*packetProc[protocol])(sessionId, packet);
}

채팅 서버에서는 OnRecv()로 받은 packet으로 데이터를 받아 함수를 호출해 처리한다.

이처럼 WorkerThread에 작업이 물려있기 때문에 packet 처리를 보다 빨리 할 수 있도록 구조를 변경했다.

✅ 수정 후

void ChattingServer::OnRecv(__int64 sessionId, Packet* packet)
{	
	Job* job = _jobPool->Alloc();	
	job->sessionId = sessionId;		

	Packet* jobPacket = Packet::Alloc();
	if (jobPacket == nullptr)
	{
		_jobPool->Free(job);
		CRASH("JobPacket nullptr");		
	}

	jobPacket->Clear();
	jobPacket->SetHeader(packet->GetBufferPtr(), sizeof(Packet::EncodeHeader));
	jobPacket->InsertData(packet->GetFrontBufferPtr(), packet->GetUseBufferSize() - sizeof(Packet::EncodeHeader));

	job->packet = jobPacket;

	EnterCriticalSection(&_jobQueueLock);
	_jobQueue.push(job);
	LeaveCriticalSection(&_jobQueueLock);

	SetEvent(_hUpdateWakeEvent);
}

위처럼 OnRecv()에 진입하면 job을 생성해 패킷을 다시 담고, jobQueue에 넣은 후, UpdateThread를 깨운다.

unsigned __stdcall ChattingServer::UpdateThreadProc(void* argument)
{
	ChattingServer* instance = (ChattingServer*)argument;
	while (1)
	{
		WaitForSingleObject(instance->_hUpdateWakeEvent, INFINITE);

		EnterCriticalSection(&instance->_jobQueueLock);
		while (!instance->_jobQueue.empty())
		{
			Job* job = instance->_jobQueue.front();
			instance->_jobQueue.pop();			

			LeaveCriticalSection(&instance->_jobQueueLock);

			short protocol;
			*job->packet >> protocol;

			(instance->*(instance->packetProc[protocol]))(job->sessionId, job->packet);

			job->packet->Free();			

			EnterCriticalSection(&instance->_jobQueueLock);
			instance->_jobPool->Free(job);
		}
		LeaveCriticalSection(&instance->_jobQueueLock);
	}
}

UpdateThread에서는 UpdateWakeEvent가 활성화될때까지 대기하다가, 일어나서 job을 처리한다.

https://github.com/YamSaeng/ChattingServer/blob/master/ChattingServer/ChattingServer.CPP

스트레스테스트를 진행하는 도중에 외국 ip가 접근하는 것을 발견했다.

클라가 접속하면 ip를 확인하고, 한국 ip가 아니면 접속을 끊는 것을 구현했다.

우선 한국 ip인지 외국 ip인지 확인해야하므로 기준 값이 필요하니까 대륙별 인터넷 레지스트리(RIR, Regional Internet registry )에서 대륙별 ip 범위를 curl을 이용해 가져온다. 가져온 ip를 활용해 서버에 접속한 클라의 ip를 검사해서 한국인지 외국인지 검사하여 외국이면 거절한다.

✅ 사용방법

curl을 이용해 RIR에 접근하여 ip 정보를 받아온다. 

클라가 접속하면 nationCode를 확인하고 LOOPBACK과 KR이 아니라면 접속을 끊는다.

https://github.com/YamSaeng/ChattingServer/blob/master/ChattingServer/IPCountryChecker.h

서버에서 관리하는 Session을 vector<>로 관리하는 것을 Lockfree Stack을 이용해 index를 관리하는 방식으로 변경했다.

vector로 관리하다보니 Sessions를 접근하거나 session을 넣거나 뺄때 Lock을 걸어줘야했는데, 매우 빈번한 작업이므로 Sessions를 일반 배열로 변경하고, index를 원자적으로 접근해 해결한다.

방식은 다음과 같다.

서버를 준비할때, 배열에 미리 Session을 할당한다.

그리고 index를 LockfreeStack에 보관한다.

클라가 서버에 접속하면 index를 하나 꺼내서 Session을 찾은 후에 초기화 해준다.

클라가 접속이 끊어지면, 

Release에서 index를 반납해서 session이 더이상 사용되지 않음을 알려준다.

채팅 서버에서 사용하는 Packet은 메시지를 전송하기 전에 할당하므로 매우 많이 동적할당하게 된다.

따라서 ObjectPool을 이용해 새로 할당받고 사용하지 않으면 반납한 후, 안에 저장중인 Packet을 대여해준다.

static ObjectPool<Packet> _objectPool; // ObjectPool을 사용하여 Packet을 할당

이처럼 static으로 선언하여 Packet 클래스가 전역으로 objectPool을 관리하도록 해준다.

// Packet을 할당
static Packet* Alloc(); 
// Packet을 반납
void Free();

Alloc을 이용해 Packet을 할당받고 Free를 이용해 반납한다.

✅ 사용방법

Packet* packet = Packet::Alloc(); // packet 할당
... // packet에 데이터담음
packet->Free(); // packet 반납

6/11

채팅 서버를 스트레스 테스트 하기 위해 더미 클라이언트를 만들었다.

사용자로부터 입력을 받아 더미 클라이언트를 생성하고 채팅서버로 접속한 후 일정 시간 마다 정해진 문자열 데이터를 서버로 전송한다. 과부하를 주기 위해 채팅 서버에서는 입력받은 데이터를 접속중인 모든 더미 클라이언트한테 전달한다. 현재는 단순히 접속하고 데이터만 전송하지만 추후에 랜덤하게 접속을 끊고 서버에 다시 접속하여 더 많은 경우를 테스트하려고 생각중이다.

50개의 더미클라이언트를 접속하고 sendPacketTPS와 recvPacketTPS를 재어보니 아래와 같았다.

✅ 더미 클라이언트 접속 끊기

더미 클라의 접속을 끊으려고 하다보니 CoreNetwork에서 관리하는 sessions가 문제가 되었다. ( vector로 관리 중 )

지금까지는 sessions를 접근할때 어차피 끊는 경우가 없기 때문에 동기화가 필요 없었는데, 이제 연결을 끊어야 하니 동기화 객체가 필요해졌다. 문제는 sessions에 접근해서 session을 찾는 FindSession 함수의 호출 횟수가 매우 잦아서 때문에, 일일이 동기화 객체를 사용하면 오버헤드가 심할 것이라고 자체판단했다. 

vector 대신 배열을 사용하고, 배열에 접근하는 index를 따로 관리하는 방법으로 변경해야겠다.

이번 채팅서버는 채팅 채널을 생성하고, 해당 채널에 유저들이 접속하고,

해당 채널에 속한 유저들끼리 채팅하는 것을 기본으로 한다. 이에 맞는 프로토콜을 구성했다.

#pragma once

enum en_CHATTING_SERVER_PACKET_TYPE
{
	en_CHATTING_SERVER_PACKET = 0,

	//=======================================
	// 하트비트 요청 packet
	// short Type
	//=======================================
	en_CHATTING_SERVER_PACKET_C2S_HEARTBEAT = 1,

	//=======================================
	// 하트비트 응답 packet
	// short Type
	//=======================================
	en_CHATTING_SERVER_PACKET_S2C_HEARTBEAT = 2,

	//=======================================
	// 채팅 요청 packet
	// short Type
	// string Message
	//=======================================
	en_CHATTING_SERVER_PACKET_C2S_CHAT = 3,

	//=======================================
	// 채팅 응답 packet
	// short Type
	// string Message
	//=======================================
	en_CHATTING_SERVER_PACKET_S2C_CHAT = 4,

	//========================================================
	// 채팅채널 생성 요청 packet
	// short Type
	// string ChannelName
	//========================================================
	en_CHATTING_SERVER_PACKET_C2S_CREATE_CHATTING_CHANNEL = 5,

	//========================================================
	// 채팅채널 생성 응답 packet
	// short Type
	// string ChannelName
	//========================================================
	en_CHATTING_SERVER_PACKET_S2C_CREATE_CHATTING_CHANNEL = 6,

	//======================================================
	// 채팅채널 입장 요청 packet
	// short Type
	// string ChannelName
	//======================================================
	en_CHATTING_SERVER_PACKET_C2S_JOIN_CHATTING_CHANNEL = 7,

	//======================================================
	// 채팅채널 입장 응답 packet
	// short Type
	// string ChannelName
	//======================================================
	en_CHATTING_SERVER_PACKET_S2C_JOIN_CHATTING_CHANNEL = 8,

	//========================================================
	// 채팅채널 나가기 요청 packet
	// short Type
	// string ChannelName
	//========================================================
	en_CHATTING_SERVER_PACKET_C2S_LEAVE_CHATTING_CHANNEL = 9,

	//========================================================
	// 채팅채널 나가기 응답 packet
	// short Type	
	//========================================================
	en_CHATTING_SERVER_PACKET_S2C_LEAVE_CHATTING_CHANNEL = 10,

	//================================================
	// 채팅채널 삭제 요청 packet
	// short Type
	// string ChannelName
	//================================================
	en_CHATTING_SERVER_PACKET_C2S_DELETE_CHANNEL = 11,

	//================================================
	// 채팅채널 삭제 응답 packet
	// short Type
	// string ChannelName
	//================================================
	en_CHATTING_SERVER_PACKET_S2C_DELETE_CHANNEL = 12,

	//=======================================================
	// 채팅채널 목록 요청 packet
	// short Type	
	//=======================================================
	en_CHATTING_SERVER_PACKET_C2S_CHATTING_CHANNEL_LIST = 13,

	//=======================================================
	// 채팅채널 목록 응답 packet
	// short Type
	// string[] ChannelList
	//=======================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_LIST = 14,

	//==============================================================
	// 채팅채널 참여자 목록 요청 packet
	// short Type	
	//==============================================================
	en_CHATTING_SERVER_PACKET_C2S_CHATTING_CHANNEL_MEMBER_LIST = 15,

	//==============================================================
	// 채팅채널 참여자 목록 응답 packet
	// short Type	
	// string[] MemberList
	//==============================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_MEMBER_LIST = 16,

	//=============================================================
	// 채팅채널 참여자 추가 요청 packet
	// short Type
	// string ChannelName
	// string MemberName
	//=============================================================
	en_CHATTING_SERVER_PACKET_C2S_CHATTING_CHANNEL_ADD_MEMBER = 17,

	//=============================================================
	// 채팅채널 참여자 추가 응답 packet
	// short Type
	// string ChannelName
	// string MemberName
	//=============================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_ADD_MEMBER = 18,

	//================================================================
	// 채팅채널 참여자 제거 요청 packet
	// short Type
	// string ChannelName
	// string MemberName
	//================================================================
	en_CHATTING_SERVER_PACKET_C2S_CHATTING_CHANNEL_REMOVE_MEMBER = 19,

	//================================================================
	// 채팅채널 참여자 제거 응답 packet
	// short Type
	// string ChannelName
	// string MemberName
	//================================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_REMOVE_MEMBER = 20,

	//=======================================================================
	// 채팅채널 참여자 상태 변경 요청 packet
	// short Type
	// string ChannelName
	// string MemberName
	// int Status (0: offline, 1: online)
	//=======================================================================
	en_CHATTING_SERVER_PACKET_C2S_CHATTING_CHANNEL_MEMBER_STATUS_CHANGE = 21,

	//=======================================================================
	// 채팅채널 참여자 상태 변경 응답 packet
	// short Type
	// string ChannelName
	// string MemberName
	// int Status (0: offline, 1: online)
	//=======================================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_MEMBER_STATUS_CHANGE = 22,

	//=================================================================
	// 채팅채널 메시지 알림 응답 packet
	// short Type
	// string ChannelName	
	// string Message
	//=================================================================
	en_CHATTING_SERVER_PACKET_S2C_CHATTING_CHANNEL_GLOBAL_MESSAGE = 23,
};

Session이 가지고 있는 SendQueue는 STL의 큐를 사용하기 때문에 멀티 쓰레드에서 사용하려면, 동기화 객체를 사용해야한다.

앞서 작성한 SendPacket, SendPost, SendComplete 모두 멀티 쓰레드 환경에서 사용하면 위험하다.

내부적으로 CriticalSection을 관리하며 Queue에 데이터를 넣고 빼는 클래스를 따로 준비했다.

#pragma once
#include"pch.h"
#include"Packet.h"

#define SEND_QUEUE

#ifdef SEND_QUEUE
#define SEND_QUEUE_DLL __declspec(dllexport)
#else
#define SEND_QUEUE_DLL __declspec(dllimport)
#endif

class SEND_QUEUE_DLL SendQueue
{
private:
	queue<Packet*> queue;
	CRITICAL_SECTION cs;
public:
	SendQueue();
	~SendQueue();

	// packet 넣기
	void Push(Packet* packet);

	// packet 빼기
	Packet* Pop();

	// packet이 있는지 확인
	bool IsEmpty();

	// 내부 queue 청소
	void Clear();

	// queue에 있는 packet의 개수 반환
	size_t Size();
};

생성자에서 CriticalSection을 초기화 하고, 소멸자에서 삭제한다.

내부 큐에 Packet을 넣고, 뺄때 CriticalSection을 이용해 동기화시켜준다.

#pragma once

#include"../Packet.h"
#include"../RingBuffer.h"
#include"../SendQueue.h"

#pragma comment(lib, "..\\x64\\Debug\\NetworkLib.lib")

struct IOBlock
{
	// IO 작업 횟수를 기록해둘 변수 
	// Session을 사용하고 있는지에 대한 여부
	LONG64 IOCount = 0;
	
	// Release를 했는지 안했는지에 대한 여부
	LONG64 IsRelease = false;
};

struct Session
{
	LONG sessionId = 0;
	SOCKET clientSocket; // 클라이언트 소켓
	
	SOCKADDR_IN clientAddr; // 서버에 접속한 클라 주소

	RingBuffer recvRingBuffer; // recvBuf
	SendQueue sendQueue; // sendQueue	

	OVERLAPPED recvOverlapped = {}; // WSARecv 통지
	OVERLAPPED sendOverlapped = {};	// WSASend 통지

	IOBlock* IOBlock = nullptr; // Session이 사용중인지, 해제되었는지 확인
	
	LONG isSend; // 세션에 대해 WSASend 작업을 하고 있는지 안하고 있는지 여부

	vector<Packet*> sendPacket; // 전송할 패킷들을 담아둠
	LONG sendPacketCount; // 전송한 패킷의 개수를 기록
};

Session 구조체에서 queue<Packet*> sendQueue 를 삭제하고 SendQueue를 추가시켰다.

Session의 sendQueue를 사용하는 모든 코드에서 새로 작성한 클래스에 맞게 수정했다.

IOCP Send 완료 통지 함수인 SendComplete를 구현했다.

WSASend 완료 통지가 오면, 전송 완료된 패킷을 먼저 정리한다.

isSend을 0으로 바꿔 WSASend를 걸 수 있도록 한다.

isSend가 0으로 바뀌기 전에 큐잉된 packet들이 있을 수 있으므로 sendQueue의 크기를 재고 0보다 크면 SendPost를 호출한다.

void CoreNetwork::SendComplete(Session* sendCompleteSession)
{
	// 전송 완료된 패킷을 정리한다.
	for (int i = 0; i < sendCompleteSession->sendPacketCount; i++)
	{
		delete sendCompleteSession->sendPacket[i];
		sendCompleteSession->sendPacket[i] = nullptr;
	}

	sendCompleteSession->sendPacketCount = 0;

	// isSend를 0 으로 바꿔서 WSASend를 걸수 있도록 해준다.
	InterlockedExchange(&sendCompleteSession->isSend, 0);

	// 만약 위에서 바꾸기 전에 큐잉만 하고 빠질 경우
	// 여기서 크기를 검사해서 WSASend를 걸어준다.
	if (sendCompleteSession->sendQueue.size() > 0)
	{
		SendPost(sendCompleteSession);
	}
}

WSASend로 packet을 전송하는 과정은 2가지 과정으로 나뉜다.

우리가 작성하는 ChattingServer에서 Session은 WSASend를 한번만 걸 수 있다.

즉, 하나의 워커 쓰레드가 WSASend를 걸면 해당 WSASend가 완료될때까지 다른 워커 쓰레드에서는 WSASend를 걸 수 없다.

SendPacket은 여러 쓰레드에서 호출할 수 있기 때문에 SendPacket에서는 packet을 큐잉을 중점으로 구현한다. ( 물론 SendPost를 호출해 WSASend 걸기를 시도 한다 )

1️⃣ SendPacket

WSASend를 걸 session을 찾고, 전송할 packet을 큐잉하고, WSASend를 건다.

void CoreNetwork::SendPacket(__int64 sessionId, Packet* packet)
{	
	Session* sendSession = FindSession(sessionId);
	if (sendSession == nullptr)
	{
		return;
	}

	packet->Encode();

	// 패킷 큐잉
	sendSession->sendQueue.push(packet);

	// WSASend 등록
	SendPost(sendSession);

	ReturnSession(sendSession);
}

2️⃣ SendPost

WSASend를 걸기전 isSend를 1로 바꾸면서 진입하고, 다른 워커쓰레드에서 isSend가 false가 될때까지 WSASend를 걸지 못하도록 막는다. 보내려고 진입했으나 sendUseSize가 0일 경우 바로나가지 않고 sendQueue의 사이즈를 다시 한번 더 확인해서 SendPost를 다시 진행한다. 

이후 전송할 packet을 WSABUF에 담고 WSASend를 건다.

void CoreNetwork::SendPost(Session* sendSession)
{
	int sendUseSize;
	int sendCount = 0;
	WSABUF sendBuf[500];

	do
	{
		// Session의 isSend를 1로 바꾸면서 진입한다.
		// 다른 워커 쓰레드가 WSASend를 하지 않도록 막는다.
		if (InterlockedExchange(&sendSession->isSend, 1) == 0)
		{
			sendUseSize = sendSession->sendQueue.size();

			if (sendUseSize == 0)
			{
				InterlockedExchange(&sendSession->isSend, 0);

				if (!sendSession->sendQueue.empty())
				{
					continue;
				}
				else
				{
					return;
				}
			}			
		}
		else
		{
			return;
		}
	} while (0);

	// session이 보내야할 packet의 개수를 지정한다.
	sendSession->sendPacketCount = sendCount = sendUseSize;

	// 보내야할 패킷의 개수만큼 sendQueue에서 패킷을 꺼내서 WSABUF에 넣는다.
	for (int i = 0; i < sendCount; i++)
	{
		if (sendSession->sendQueue.size() == 0)
		{
			break;
		}

		InterlockedIncrement(&_sendPacketTPS);

		Packet* packet = sendSession->sendQueue.front();
		sendSession->sendQueue.pop();
		
		sendBuf[i].buf = packet->GetHeaderBufferPtr();
		sendBuf[i].len = packet->GetUseBufferSize();

		sendSession->sendPacket[i] = packet;
	}

	// WSAsend를 걸기 전에 한번 청소해준다.
	memset(&sendSession->sendOverlapped, 0, sizeof(OVERLAPPED));

	// IOCount를 1 증가시켜 해당 session이 release 대상이 되지 않도록 한다.
	InterlockedIncrement64(&sendSession->IOBlock->IOCount);
		
	int WSASendRetval = WSASend(sendSession->clientSocket, sendBuf, sendCount, NULL, 0, (LPWSAOVERLAPPED)&sendSession->sendOverlapped, NULL);
	if (WSASendRetval == SOCKET_ERROR)
	{
		DWORD error = WSAGetLastError();
		if (error != ERROR_IO_PENDING)
		{
			if (InterlockedDecrement64(&sendSession->IOBlock->IOCount) == 0)
			{
				ReleaseSession(sendSession);
			}
		}
	}
}

2025.06.01 - [ChattingServer] - [ChattingServer] RingBuffer ( 원형 큐 )

앞서 포스팅한 RingBuffer( 원형 큐 )를 이용해 클라이언트로부터 데이터를 받는다.

원형큐이므로 연속으로 한번에 데이터를 넣을 수 있는 공간과 전체 남은 공간을 비교해서 링버퍼가 2개로 나뉘어있는지 확인해 WSABUF를 구성한다.

IOCount를 증가 시켜 지금 session이 사용중임을 기록하고 WSARecv를 건다.

void CoreNetwork::RecvPost(Session* recvSession, bool isAcceptRecvPost)
{
	int recvBuffCount = 0;
	WSABUF recvBuf[2];

	// recvRingBuffer에 한번에 넣을 수 있는 크기를 읽는다.
	int directEnqueueSize = recvSession->recvRingBuffer.GetDirectEnqueueSize();
	// 남아 있는 recvRingBuffer의 크기를 읽는다.
	int recvRingBuffFreeSize = recvSession->recvRingBuffer.GetFreeSize();

	// 만약 남아 있는 recvRingBuffer의 크기가 한번에 넣을 수 있는 크기보다 크다면
	// recvRingBuffer가 2개의 공간으로 나뉘어 있는 것을 확인할 수 있다.
	if (recvRingBuffFreeSize > directEnqueueSize)
	{
		recvBuffCount = 2;
		recvBuf[0].buf = recvSession->recvRingBuffer.GetRearBufferPtr();
		recvBuf[0].len = directEnqueueSize;

		recvBuf[1].buf = recvSession->recvRingBuffer.GetBufferPtr();
		recvBuf[1].len = recvRingBuffFreeSize - directEnqueueSize;
	}
	else
	{
		recvBuffCount = 1;
		recvBuf[0].buf = recvSession->recvRingBuffer.GetRearBufferPtr();
		recvBuf[0].len = directEnqueueSize;
	}

	// WSARecv를 걸기 전에 한번 청소해준다.
	memset(&recvSession->recvOverlapped, 0, sizeof(OVERLAPPED));
		
	if (isAcceptRecvPost == false)
	{
		InterlockedIncrement64(&recvSession->IOBlock->IOCount);
	}	

	DWORD flags = 0;

	int WSARecvRetval = WSARecv(recvSession->clientSocket, recvBuf, recvBuffCount, NULL, &flags, (LPWSAOVERLAPPED)&recvSession->recvOverlapped, NULL);
	if (WSARecvRetval == SOCKET_ERROR)
	{
		DWORD error = WSAGetLastError();
		if (error != ERROR_IO_PENDING)
		{
			if (InterlockedDecrement64(&recvSession->IOBlock->IOCount) == 0)
			{
				ReleaseSession(recvSession);
			}
		}
	}
}