[C++] Session

[C++과 언리얼로 만드는 MMORPG 게임 개발 시리즈] Part4: 게임 서버를 참고하여 정리한 내용입니다.

Listener

IocpObject를 상속받는 클래스이다. 여기서 Session이 Accept하는 작업을 관리한다.

Listener.h

#pragma once
#include "IocpCore.h"
#include "NetAddress.h"

class AcceptEvent;

/*
	Listener
*/

class Listener : public IocpObject
{
public:
	Listener() = default;
	~Listener();

public:
	/* 외부에서 사용 */
	bool StartAccept(NetAddress netAddress);
	void CloseSocket();

public:
	/* 인터페이스 구현 */
	// IocpObject을(를) 통해 상속됨
	virtual HANDLE GetHandle() override;
	virtual void Dispatch(IocpEvent* iocpEvent, int32 numOfBytes) override;

private:
	/* 수신 관련 */
	void RegisterAccept(AcceptEvent* acceptEvent);
	void ProcessAccept(AcceptEvent* acceptEvent);

protected:
	SOCKET _socket = INVALID_SOCKET;
	Vector<AcceptEvent*> _acceptEvents;
};

Listener.cpp

#include "pch.h"
#include "Listener.h"
#include "SocketUtils.h"
#include "IocpEvent.h"
#include "Session.h"


/*
	Listener
*/

Listener::~Listener()
{
	SocketUtils::Close(_socket);
	for(AcceptEvent* acceptEvent : _acceptEvents)
	{
		xdelete(acceptEvent);
	}
}

bool Listener::StartAccept(NetAddress netAddress)
{
	_socket = SocketUtils::CreateSocket();
	if (_socket == INVALID_SOCKET)
		return false;

	if (GIocpCore.Register(this) == false)
		return false;

	if (SocketUtils::SetReuseAddress(_socket, true) == false)
		return false;

	if(SocketUtils::Bind(_socket, netAddress) == false)
		return false;

	if (SocketUtils::Listen(_socket) == false)
		return false;

	const int32 acceptCount = 1;
	for (int32 i = 0; i < acceptCount; i++)
	{
		AcceptEvent* acceptEvent = xnew<AcceptEvent>();
		_acceptEvents.push_back(acceptEvent);
		RegisterAccept(acceptEvent);
	}

	return false;
}

void Listener::CloseSocket()
{
	SocketUtils::Close(_socket);
}

HANDLE Listener::GetHandle()
{
	return reinterpret_cast<HANDLE>(_socket);
}

void Listener::Dispatch(IocpEvent* iocpEvent, int32 numOfBytes)
{
	// accept가 아닌 다른 overlapped가 왔다면 에러
	// accpet가 맞다면, 정상적으로 처리
	ASSERT_CRASH(iocpEvent->GetType() == EventType::Accept);
	AcceptEvent* accpetEvent = static_cast<AcceptEvent*>(iocpEvent);
	ProcessAccept(accpetEvent);
}

void Listener::RegisterAccept(AcceptEvent* acceptEvent)
{
	Session* session = xnew<Session>();
	acceptEvent->Init();
	acceptEvent->SetSession(session);

	DWORD bytesReceived = 0;
	if (false == SocketUtils::AcceptEx(_socket, session->GetSocket(), session->_recvBuffer, 0, sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16, OUT & bytesReceived, static_cast<LPOVERLAPPED>(acceptEvent)))
	{
		// 실패했을 때
		int32 errCode = WSAGetLastError();
		if (errCode != WSA_IO_PENDING)
		{
			// Pending 상태 외의 에러코드라면, 문제가 있는 상황임
			// 일단 다시 Accept 해줌
			RegisterAccept(acceptEvent);
		}
	}
}

void Listener::ProcessAccept(AcceptEvent* acceptEvent)
{
	Session* session = acceptEvent->GetSession();

	if (false == SocketUtils::SetUpdateAcceptSocket(session->GetSocket(), _socket))
	{
		RegisterAccept(acceptEvent);
		return;
	}

	SOCKADDR_IN sockAddress;
	int32 sizeOfSockAddr = sizeof(sockAddress);
	if (SOCKET_ERROR == ::getpeername(session->GetSocket(), OUT reinterpret_cast<SOCKADDR*>(&sockAddress), &sizeOfSockAddr))
	{
		RegisterAccept(acceptEvent);
		return;
	}
	session->SetNetAddress(NetAddress(sockAddress));

	cout << "Client Conneted !" << endl;

	RegisterAccept(acceptEvent);


}

StartAccept : 클라이언트가 대상 Address에 접속할 때 호출한다.
CloseSocket : 접속을 끊고 소켓 정보를 소멸시킨다.

RegisterAccept : 이곳에서 Accept이벤트를 조립하고 IocpHandle에 등록한다.
ProcessAccept : Dispatch로부터 결과를 받아와서 처리한다. 여기서 제대로 등록되지 않았다면 RegisterAccept를 호출한다.

Session

정말 중요한 Session이다. 서버 혹은 클라이언트 소켓의 대리인이라고 보면 된다. 모든 Overlapped I/O를 처리하는 IocpObject이다.

Session.h

#pragma once
#include "IocpCore.h"
#include "IocpEvent.h"
#include "NetAddress.h"
#include "RecvBuffer.h"

class Service;

/*
	Session
*/

// 클라이언트의 정보를 담고 있는 클래스
class Session : public IocpObject
{
	// 이 친구들만 사용할 수 있도록 열어줌.
	friend class Listener;
	friend class IocpCore;
	friend class Service;

	enum
	{
		BUFFER_SIZE = 0x10000, // 64KB
	};

public:
	Session();
	virtual ~Session();

public:
						/* 외부에서 사용 */
	void				Send(BYTE* buffer, int32 len);
	bool				Connect();
	void				Disconnect(const WCHAR* cause);	// 강제 연결종료

	shared_ptr<Service> GetService() { return _service.lock(); }
	void				SetService(shared_ptr<Service> service) { _service = service; }

public:
						/* 정보 관련 */
	void				SetNetAddress(NetAddress address) { _netAddress = address; }
	NetAddress			GetAddress() { return _netAddress; }
	SOCKET				GetSocket() { return _socket; }
	bool				IsConnected() { return _connected; }
	SessionRef			GetSessionRef() { return static_pointer_cast<Session>(shared_from_this()); }

private:
						/* 인터페이스 구현 */
						// IocpObject을(를) 통해 상속됨
	virtual HANDLE		GetHandle() override;
	virtual void		Dispatch(IocpEvent* iocpEvent, int32 numOfBytes) override;

private:
						/* 전송 관련 */
	bool				RegisterConnect();
	bool				RegisterDisconnect();
	void				RegisterRecv();
	void				RegisterSend(SendEvent* sendEvent);
	
	void				ProcessConnect();
	void				ProcessDisconnect();
	void				ProcessRecv(int32 numOfBytes);
	void				ProcessSend(SendEvent* sendEvent, int32 numOfBytes);
	
	void				HandleError(int32 errorCode);

protected:
						/* 컨텐츠 코드에서 오버로딩 */
	virtual void		OnConnected() { }
	virtual int32		OnRecv(BYTE* buffer, int32 len) { return len; }
	virtual void		OnSend(int32 len) { }
	virtual void		OnDisconnected() { }

private:
	weak_ptr<Service>	_service;
	SOCKET			_socket = INVALID_SOCKET;
	NetAddress		_netAddress = {};
	Atomic<bool>	_connected = false;

private:
	USE_LOCK;
	
						/* 수신 관련 */
	RecvBuffer			_recvBuffer;


	/* 송신 관련 */

private:
						/* IocpEvent 재사용 */
	ConnectEvent		_connectEvent;
	DisconnectEvent		_disconnectEvent;
	RecvEvent			_recvEvent;
};

Session.cpp

#include "pch.h"
#include "Session.h"
#include "SocketUtils.h"
#include "Service.h"

/*
	Session
*/

Session::Session()
{
	_socket = SocketUtils::CreateSocket();
}

Session::~Session()
{
	SocketUtils::Close(_socket);
}

void Session::Send(BYTE* buffer, int32 len)
{
	// 생각해야될 점
	// 1)버퍼 관리
	// 2) sendEvent 관리? WSASend 중첩 가능성(귓속말이라던지, 몬스터 정보라던지 등 동시에 여러 개가 send될수도)

	// TEMP
	SendEvent* sendEvent = xnew<SendEvent>();
	sendEvent->owner = shared_from_this();
	sendEvent->buffer.resize(len);
	::memcpy(sendEvent->buffer.data(), buffer, len);

	WRITE_LOCK;
	RegisterSend(sendEvent);
}

// 서버끼리 연결해야 될 수도 있기 때문에 구현.
bool Session::Connect()
{
	return RegisterConnect();
}

void Session::Disconnect(const WCHAR* cause)
{
	if (_connected.exchange(false) == false)
		return;

	wcout << "Disconnected : " << cause << endl;

	OnDisconnected();	// 컨텐츠 코드에서 재정의
	SocketUtils::Close(_socket);
	GetService()->ReleaseSession(GetSessionRef());

	RegisterDisconnect();
}

HANDLE Session::GetHandle()
{
	return reinterpret_cast<HANDLE>(_socket);
}

void Session::Dispatch(IocpEvent* iocpEvent, int32 numOfBytes)
{
	switch (iocpEvent->eventType)
	{
	case EventType::Connect:
		ProcessConnect();
		break;
	case EventType::Disconnect:
		ProcessDisconnect();
		break;
	case EventType::Recv:
		ProcessRecv(numOfBytes);
		break;
	case EventType::Send:
		ProcessSend(static_cast<SendEvent*>(iocpEvent), numOfBytes);
		break;

	default:
		break;
	}
}

bool Session::RegisterConnect()
{
	if (IsConnected())
		return false;

	if (GetService()->GetSerivceType() != ServiceType::Client)
		return false;

	if (SocketUtils::SetReuseAddress(_socket, true) == false)
		return false;

	if (SocketUtils::BindAnyAddress(_socket, 0) == false)
		return false;


	_connectEvent.Init();
	_connectEvent.owner = shared_from_this();

	DWORD numOfBytes = 0;
	SOCKADDR_IN sockAddr = GetService()->GetNetAddress().GetSockAddr();
	if (SocketUtils::ConnectEx(_socket, reinterpret_cast<SOCKADDR*>(&sockAddr), sizeof(sockAddr), nullptr, 0, &numOfBytes, &_connectEvent) == false)
	{
		int errorCode = WSAGetLastError();
		if (errorCode != WSA_IO_PENDING)
		{
			_connectEvent.owner = nullptr;	// Release_ref
			return false;
		}

		return true;
	}
}

bool Session::RegisterDisconnect()
{
	_disconnectEvent.Init();
	_disconnectEvent.owner = shared_from_this();

	if (false == SocketUtils::DisconnectEx(_socket, &_disconnectEvent, TF_REUSE_SOCKET, 0))
	{
		int32 errorCode = WSAGetLastError();
		if (errorCode != WSA_IO_PENDING)
		{
			_disconnectEvent.owner = nullptr;	// Release_ref
			return false;
		}
	}
	return true;
}

void Session::RegisterRecv()
{
	if (IsConnected() == false)
		return;

	_recvEvent.Init();						// 초기화
	_recvEvent.owner = shared_from_this();	// Add_Ref

	WSABUF wsaBuf;
	wsaBuf.buf = reinterpret_cast<char*>(_recvBuffer);
	wsaBuf.len = len32(_recvBuffer);

	DWORD numOfBytes = 0;
	DWORD flags = 0;

	if (SOCKET_ERROR == ::WSARecv(_socket, &wsaBuf, 1, OUT & numOfBytes, OUT & flags, &_recvEvent, nullptr))
	{
		int32 errorCode = WSAGetLastError();
		if (errorCode != WSA_IO_PENDING)
		{
			HandleError(errorCode);
			_recvEvent.owner = nullptr;		// Release_Ref
		}
	}
}

void Session::RegisterSend(SendEvent* sendEvent)
{
	if (IsConnected() == false)
		return;

	WSABUF wsaBuf;
	wsaBuf.buf = (char*)sendEvent->buffer.data();
	wsaBuf.len = (ULONG)sendEvent->buffer.size();

	DWORD numOfBytes = 0;
	// WSASend는 여러 버퍼를 모아두고 한번에 보내도 됨. 2~3번째 인자가 그 역할을 함
	if (SOCKET_ERROR == ::WSASend(_socket, &wsaBuf, 1, OUT & numOfBytes, 0, sendEvent, nullptr))
	{
		int32 errorCode = ::WSAGetLastError();
		if (errorCode != WSA_IO_PENDING)
		{
			HandleError(errorCode);
			sendEvent->owner = nullptr;
			xdelete(sendEvent);
		}
	}

}

void Session::ProcessConnect()
{
	_connectEvent.owner = nullptr;	// Release_ref
	_connected.store(true);

	// 세션 등록
	GetService()->AddSession(GetSessionRef());

	// 컨텐츠코드에서 오버로딩
	OnConnected();

	// 수신 등록
	RegisterRecv();
}

void Session::ProcessDisconnect()
{
	_disconnectEvent.owner = nullptr;	// Release_ref
}

void Session::ProcessRecv(int32 numOfBytes)
{

	// 이거 안 해주면 Session이 소멸이 안됨.
	_recvEvent.owner = nullptr;		// Release_Ref

	// Recv 실패했을 경우
	if (numOfBytes == 0)
	{
		Disconnect(L"Recv 0");
		return;
	}

	// 컨텐츠 코드에서 재정의
	OnRecv(_recvBuffer, numOfBytes);

	// 수신 등록
	RegisterRecv();
}

void Session::ProcessSend(SendEvent* sendEvent, int32 numOfBytes)
{
	sendEvent->owner = nullptr;	//Release_Ref
	xdelete(sendEvent);

	if (numOfBytes == 0)
	{
		Disconnect(L"Send 0");
		return;
	}

	OnSend(numOfBytes);
}

void Session::HandleError(int32 errorCode)
{
	switch (errorCode)
	{
	case WSAECONNRESET:
	case WSAECONNABORTED:
		Disconnect(L"HandleError");
		break;
	default:
		// TODO : Log
		cout << "Handle Error : " + errorCode << endl;
		break;
	}
}

내용이 굉장히 많다. 메소드 기준으로 보자면,

Send : SendEvent를 조립하는 메소드이다. 이벤트를 등록하기 전, 어떤 데이터를 전달할 지를 SendEvent의 멤버변수에 버퍼데이터를 복사한다.
Connect : 접속 이벤트를 등록한다. 자세한건 RegisterEvent를 보자.
Disconnect : 접속종료를 요청하고 이벤트를 등록한다.

Dispatch : I/O처리가 끝나고 결과를 전달받는 추상 메소드이다.

RegisterConnect : Iocp에 접속이벤트를 등록한다.
RegsiterDisconnect : Iocp에 접속종료 이벤트를 등록한다.
RegisterRecv : Iocp에 수신이벤트를 등록한다.
RegisterSend : Iocp에 송신이벤트를 등록한다. 이 때, 전달할 데이터는 SendEvent의 멤버변수에 있다.

ProcessConnect : 접속 결과를 처리하는 곳이다. 접속에 성공한 세션을 리스트에 추가한다. 이제 세션한테 데이터를 받아야 하니 수신이벤트도 겸사겸사 등록해준다.
ProcessDisconnect : 딱히 할 것은 없다. 이벤트 소멸을 하지 않으면 세션이 죽지 않으므로 이정도만 해준다.
ProcessRecv : 컨텐츠 단으로 받은 데이터를 전달한다. 계속 받아야하므로 수신이벤트를 등록한다.
ProcessSend : 컨텐츠 단으로 보낸 데이터를 전달한다. 멤버변수에 버퍼(=가변데이터 <= 메모리)가 있으므로 해제해줘야한다.

Process 함수들에 공통적으로 //Release_Ref라는 주석이 적혀있는데, 이는 I/O 작업이 끝나고 돌아온 이벤트의 멤버변수인 owner, 즉 Session과의 연결고리를 끊어준다.
이전 IocpEvent에서 언급했듯이, owner라는 shared_ptr을 통해 I/O 처리가 끝나기 전에 세션이 소멸되는 것을 방지한다. 하지만, I/O가 끝나고 owner를 끊어주지 않으면 나중에 Session이 죽지 않을수도 있다. 그러니 볼 일을 다 봤다면 해제시켜주도록 하자.

Service에 대해서는 다음 글에서 다루도록 하겠다.