27. IOCP accept, recv까지 구현

이전에 NetAddress까지 만들었는데, 다음 목표로 우선 IOCP로 싱글스레드 채팅서버를 만들어보려 한다.

끝까지 이번 게시글에서는 accept와 recv까지 구현했다. IOCP의 플로우 대로 구현해봤다.

---

 

I/O Multiplexing IOCP 에코서버 만들기 with.C/C++

저번에는 IOCP로 싱글스레드 멀티플랙싱 에코서버를 만들었을 때 IOCP 서버의 동작순서를 정리했었는데 이걸 기반으로 각각의 과정을 누가 수행해야할지 생각해보고 클래스들을 구성해보자.

1. WSA 초기화 = SocketUtils에서 하는게 적절할듯 함. service start 제일 처음에 호출해야한다.
2. IOCP 핸들 생성 = IocpCore의 생성자. 생성자는 Service.Start에서 호출하자.
3. listen 소켓 생성, bind, listen = Listener 클래스 만들어서 담당.
4. listen 소켓을 IOCP 핸들에 감시 등록 = 감시 등록함수는 IocpCore에 있어야할듯 하고, 호출은 service가 하자.
5. AcceptEx로 비동기 accept 요청. = Listener의 StartAccept 함수에서 하고, service가 호출하자
   기존 accept와 다른점은 함수가 클라이언트 소켓을 리턴하지 않고, 클라이언트 소켓을 미리 만들어서 넣어줘야 한다.
6. 그리고 GetQueuedCompletionStatus에서 완료통지 대기함. = IocpCore의 Dispatch 함수에서 함. 워커스레드에서 호출. 일단은 싱글스레드로 해볼거니까 service start 안에서 마지막에 하거나 main 마지막에 하자.
7. 누군가 Connect를 하면 커널이 accept를 마치고 완료통지를 함.
8. 대기중이던 GetQueuedCompletionStatus에서 깨어남. Overlapped(IocpEvent)를 확인하니 accept 요청의 완료 통지임을 확인. = IocpCore.Dispatch에서 IocpEvent->owner (여기서는 Listener) 의 Dispatch 함수 호출해야함
9. 새 클라이언트가 들어왔으니 해당 클라이언트에 대해 WSARecv로 읽기 요청. 그리고 또 accpet도 계속 해야하니 listen 소켓에 대해서도 AcceptEx 요청. accept는 5~9 루프를 돌며 계속 받아줄 수 있음.
   = Listener의 Dispatch에서 ProcessAccept 호출해서 처리. 여기서 accept된 Session을 만들어 초기화해주고 IOCP에도 등록해줘야됨. 이 세션의 버퍼를 WSArecv에 집어넣어야함.
10. 클라이언트에게 메시지가 와서 recv 작업 마치면 GetQueuedCompletionStatus에서 Recv요청의 완료통지 확인.
    = IocpCore.Dispatch에서 IocpEvent->owner->Dispatch(). session의 dispatch에서 switch 문으로 ProcessRecv 처리. 채팅서버니까 일단 다른 모든 세션의 writebuffer에 넣어두자. (비효율적인데 나중에 생각하자.)
11. recv에 온 데이터를 그대로 클라이언트에게 WSASend로 send 비동기 요청 함.
    = Session에 RegisterSend를 만들어 요청하자.
12. ...

---

Service

서버 관리자.

메인문에서 생성하고 사용함.

int main()
{
	ServiceRef service = make_shared<Service>(NetAddress("0.0.0.0", 9000));
	service->Start();
}

class Service : public std::enable_shared_from_this<Service>
{
public:
	Service(NetAddress listenerAddr);
	~Service();

	void Start();

	IocpCoreRef GetIocpCore() { return _iocpCore; }
	NetAddress GetListenerAddr() { return _listenerAddr; }
private:
	IocpCoreRef _iocpCore;
	NetAddress _listenerAddr;
};

Service::Service(NetAddress listenerAddr) : _listenerAddr(listenerAddr)
{
	_iocpCore = make_shared<IocpCore>();
}

Service::~Service()
{
}

void Service::Start()
{
	SocketUtils::Init();

	ListenerRef listener = make_shared<Listener>(shared_from_this());

	listener->StartAccept();

	while (true)
	{
		_iocpCore->Dispatch();
	}
}

SocketUtils::Init으로 WSA 초기화 해준 뒤에 IocpCore와 Listenr를 만들어서 서버를 돌리는 역할이다.

 

IocpCore

IOCP 핸들 생성과 보유 관리, 소켓을 IOCP에 등록하는 클래스.

Dispatch 함수에서 GetQueuedCompletionStatus 에서 대기하다가 완료통지를 받으면 깨어나서 발생한 이벤트 owner의 Dispatch함수를 실행시켜서 네트워크 작업을 처리해줌.

class IocpCore
{
public:
	IocpCore();
	~IocpCore();

	bool Dispatch();

	bool RegisterHandle(HANDLE newHandle);
private:
	HANDLE _iocpHandle;
};

IocpCore::IocpCore()
{
	_iocpHandle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
	ASSERT_CRASH(_iocpHandle != INVALID_HANDLE_VALUE);
}

IocpCore::~IocpCore()
{
	::CloseHandle(_iocpHandle);
}

bool IocpCore::Dispatch()
{
	DWORD numOfBytes = 0;
	ULONG_PTR completionKey;
	IocpEvent* event = nullptr;

	bool result = GetQueuedCompletionStatus(_iocpHandle, &numOfBytes, &completionKey,
		reinterpret_cast<LPOVERLAPPED*>(&event), INFINITE);
	if (result != 0) // 정상적으로 이벤트가 발생한 경우
	{
		IocpObjectRef owner = event->GetOwner();
		owner->Dispatch(numOfBytes, event);
	}
	else // 오류가 발생한 경우
	{

	}
	return true;
}

bool IocpCore::RegisterHandle(HANDLE newHandle)
{
	if (CreateIoCompletionPort(newHandle, _iocpHandle, 0, 0) == nullptr)
		return false;
	return true;
}

 

IocpEvent

Overlapped를 상속받아서 비동기IO 함수 쓸 때 포인터로 첨부한다. 나중에 첨부된 포인터를 그대로 완료통지받을때 받아 써야하니까 스코프 나갈때 없어지지 않도록 동적할당 해줘야한다.

owner = 발생할 이벤트를 처리할 주체

EventType = 무슨 이벤트인지 (accept, recv, send 등)

 

각 이벤트별로 하위클래스도 만들어줬다.

acceptEvent의 경우에 owner인 listener 뿐 아니라 초기화 해줄 Session과 accept할 때 buffer로 받는 addr정보도 같이 저장해야 하기때문에 만들었다.

#pragma once

enum class EventType : uint8
{
	Connect,
	Disconnect,
	Accept,
	Recv,
	Send
};

class IocpEvent : public OVERLAPPED
{
public:
	IocpEvent(IocpObjectRef iocpObject, EventType eventType);
	EventType GetEventType() { return _eventType; }
	IocpObjectRef GetOwner() { return _owner; }
private:
	void Init();
	IocpObjectRef _owner;
	EventType _eventType;
};

class AcceptEvent : public IocpEvent
{
public:
	AcceptEvent(IocpObjectRef iocpObject) : IocpEvent(iocpObject, EventType::Accept) {}
	void SetSession(SessionRef session) { _session = session; }
	SessionRef GetSession() { return _session; }
	BYTE* GetAcceptBuffer() { return _acceptBuffer; }
private:
	SessionRef _session;
	BYTE _acceptBuffer[(sizeof(SOCKADDR_IN) + 16) * 2] = {0};
};

class RecvEvent : public IocpEvent
{
public:
	RecvEvent(IocpObjectRef iocpObject) : IocpEvent(iocpObject, EventType::Recv) {}
};

 

IocpObject

Listener, Session 같이 Iocp에 등록되어서 통지받을 애들의 부모클래스.

dispatch를 호출하면 오버라이딩된 Listener나 Session의 Dispatch가 호출되어 서로다른 동작을 할 수 있게 하기 위해 만들어진 완전추상클래스 인터페이스.

class IocpObject : public enable_shared_from_this<IocpObject>
{
public:
	virtual HANDLE GetHandle() abstract;
	virtual void Dispatch(int32 numOfBytes, IocpEvent* event) abstract;
};

 

Listener

listen소켓 담당자.

RegisterAccept, ProcessAccept로 새 클라이언트 받고, 초기화해주는 클래스.

class Listener : public IocpObject
{
public:
	Listener(ServiceRef service);
	~Listener();
public:
	virtual HANDLE GetHandle() override { return (HANDLE)_listenSocket; }
	virtual void Dispatch(int32 numOfBytes, IocpEvent* event) override;

public:
	NetAddress GetAddress() { return _address; }
public:
	bool StartAccept();

	void RegisterAccept(AcceptEvent* acceptEvent);
	void ProcessAccept(SessionRef session, AcceptEvent* acceptEvent);

private:
	SOCKET _listenSocket = INVALID_SOCKET;
	ServiceRef _service;
	NetAddress _address;
};

Listener::Listener(ServiceRef service) : _service(service), _address(service->GetListenerAddr())
{
	_listenSocket = SocketUtils::CreateSocket();
	if (_listenSocket == INVALID_SOCKET)
		CRASH("Failed to create listen socket");
}

Listener::~Listener()
{
	SocketUtils::CloseSocket(_listenSocket);
}

void Listener::Dispatch(int32 numOfBytes, IocpEvent* event)
{
	if (event->GetEventType() != EventType::Accept)
		CRASH("Invalid event type for Listener");
	AcceptEvent* acceptEvent = static_cast<AcceptEvent*>(event);
	SessionRef session = acceptEvent->GetSession();

	ProcessAccept(session, acceptEvent);
	RegisterAccept(acceptEvent);
}

bool Listener::StartAccept()
{
	if (SocketUtils::SetReuseAddress(_listenSocket, true) == false)
		CRASH("Failed to set reuse listen socket");

	if (SocketUtils::SetTcpNoDelay(_listenSocket, true) == false)
		CRASH("Failed to set nodelay listen socket");

	if (SocketUtils::BindSocket(_listenSocket, _address) == false)
		CRASH("Failed to bind listen socket");

	if (SocketUtils::ListenSocket(_listenSocket, SOMAXCONN) == false)
		CRASH("Failed to listen on listen socket");

	if (_service->GetIocpCore()->RegisterHandle(GetHandle()) == false)
		CRASH("Failed to register listen socket to IOCP");

	AcceptEvent* acceptEvent = new AcceptEvent(shared_from_this());
	if (acceptEvent == nullptr)
		CRASH("Failed to create accept event");

	RegisterAccept(acceptEvent);

	return true;
}

void Listener::RegisterAccept(AcceptEvent* acceptEvent)
{
	SOCKET clientSocket = SocketUtils::CreateSocket();
	if (clientSocket == INVALID_SOCKET)
		return;

	SessionRef session = make_shared<Session>(clientSocket);
	if (acceptEvent == nullptr)
		CRASH("Failed to create accept event");
	acceptEvent->SetSession(session);

	if (false == SocketUtils::AcceptEx(_listenSocket, clientSocket, acceptEvent->GetAcceptBuffer(), 0,
		sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16, NULL, static_cast<LPOVERLAPPED>(acceptEvent)))
	{
		if (WSAGetLastError() != ERROR_IO_PENDING)
		{
			//TODO 적절한 처리
		}
	}
}

void Listener::ProcessAccept(SessionRef session, AcceptEvent* acceptEvent)
{
	if (SocketUtils::SetUpdateAcceptSocket(session->_socket, _listenSocket) == false)
		return;

	if (session->SetAddressFromAcceptBuffer(acceptEvent->GetAcceptBuffer()) == false)
		return;

	if (_service->GetIocpCore()->RegisterHandle(session->GetHandle()) == false)
		return;

	RecvEvent* recvEvent = new RecvEvent(session);
	if (recvEvent == nullptr)
		return;

	session->RegisterRecv(recvEvent);
}

RegisterAccept의 AcceptEx의 세번째 인자로 acceptEvent의 버퍼를 넣는데, 나중에 GetQueuedCompletionStatus로 완료통지 받으면 버퍼에 local(나. 서버)과 remote(상대. 클라이언트)정보가 바이트 배열로 주어지는데, GetAcceptExSockaddrs 함수를 사용해 버퍼에 적힌 내용을 SOCKADDR에 담을 수 있다.

 

Session

client 소켓 담당자.

주로 Recv, Send 를 처리할 예정인 클래스. 

class Session : public IocpObject
{
	friend class Listener;
	enum { BUFFER_SIZE = 1024 };
public:
	Session(SOCKET socket);
	virtual ~Session();
public:
	virtual HANDLE GetHandle() override { return (HANDLE)_socket; }
	virtual void Dispatch(int32 numOfBytes, IocpEvent* event) override;

public:
	void RegisterRecv(IocpEvent* recvEvent);
	void ProcessRecv(int32 numOfBytes, RecvEvent* recvEvent);

	bool SetAddressFromAcceptBuffer(BYTE* buffer);

	BYTE* GetRecvBuffer() { return _recvBuffer; }
	BYTE* GetSendBuffer() { return _sendBuffer; }
private:
	SOCKET _socket = INVALID_SOCKET;
	NetAddress _address;

	BYTE _recvBuffer[BUFFER_SIZE];
	BYTE _sendBuffer[BUFFER_SIZE];
};

Session::Session(SOCKET socket) : _socket(socket) {}

Session::~Session()
{
	if (_socket != INVALID_SOCKET)
		SocketUtils::CloseSocket(_socket);
}

void Session::Dispatch(int32 numOfBytes, IocpEvent* event)
{
	
	switch (event->GetEventType())
	{
	case EventType::Connect:
		break;
	case EventType::Disconnect:
		break;
	case EventType::Send:
		break;
	case EventType::Recv:
		ProcessRecv(numOfBytes, reinterpret_cast<RecvEvent*>(event));
		break;
	default:
		return;
	}
}

void Session::RegisterRecv(IocpEvent* recvEvent)
{
	WSABUF wsaBuf;
	wsaBuf.buf = reinterpret_cast<char*>(_recvBuffer);
	wsaBuf.len = BUFFER_SIZE;

	DWORD numOfBytes = 0;
	DWORD flags = 0;

	if (SOCKET_ERROR == WSARecv(_socket, &wsaBuf, 1, OUT & numOfBytes, &flags,
		static_cast<OVERLAPPED*>(recvEvent), nullptr))
	{
		if (WSAGetLastError() != WSA_IO_PENDING)
		{
			//TODO 적절한 처리
		}
	}
}

void Session::ProcessRecv(int32 numOfBytes, RecvEvent* recvEvent)
{
	cout << "Received " << numOfBytes << " bytes from client" << endl;
	::memcpy(_sendBuffer, _recvBuffer, numOfBytes);
	_sendBuffer[numOfBytes] = '\0';
	cout << _sendBuffer << endl;
	RegisterRecv(recvEvent);
}

bool Session::SetAddressFromAcceptBuffer(BYTE* buffer)
{
	SOCKADDR_IN* serverAddr = nullptr;
	SOCKADDR_IN* clientAddr = nullptr;
	int32 serverAddrLen = 0;
	int32 clientAddrLen = 0;
	SocketUtils::GetAcceptExSockaddrs(
		buffer,
		0,
		sizeof(SOCKADDR_IN) + 16,
		sizeof(SOCKADDR_IN) + 16,
		(SOCKADDR**)&serverAddr, &serverAddrLen,
		(SOCKADDR**)&clientAddr, &clientAddrLen
	);
	if (serverAddr == nullptr || clientAddr == nullptr)
		return false;
	_address.SetAddr(*clientAddr);
	return true;
}

일단은 ProcessRecv에서 recvBuffer의 내용을 sendBuffer에 담고, 출력해보고있다. writeBuffer를 다른 유저들에게 보내는 기능은 다음에 구현해볼 것.

 

SocketUtils

GetAcceptExSockaddrs 사용하기위해 관련된 부분 추가됨.

ASSERT_CRASH(LoadExtensionFunction(WSAID_GETACCEPTEXSOCKADDRS,
	reinterpret_cast<LPVOID*>(&GetAcceptExSockaddrs)) == 0);

 

NetAddress

port가 uint16길래 GetPort의 리턴값을 바꿔줬다.

기본생성자, SOCKADDR_IN을 인자로 받는 생성자를 추가했다. Setter도 추가했다.

class NetAddress
{
public:
	NetAddress() = default;
	NetAddress(string ip, int port);
	NetAddress(const NetAddress& addr);
	NetAddress(SOCKADDR_IN addr);
	~NetAddress();

	void SetAddr(const NetAddress& addr) { _sockAddr = addr._sockAddr; }

	const SOCKADDR_IN&	GetAddr() { return _sockAddr; }
	string				GetIp();
	uint16				GetPort() { return ::ntohs(_sockAddr.sin_port); }
private:
	SOCKADDR_IN _sockAddr;
};

 

recv까지 잘 된다!

recv까지는 잘 되는걸 확인했으니 다음에는send로 서버에 접속한 모든 클라이언트에게 메시지를 전송시켜보자.

---

여기까지의  git

Refactor: accept 시 sockaddr 버퍼관리자 session -> acceptevent 로변경 · Dodontak/Project_Island_GameServer@8da336f