42. JobQueue(1)

지난번에는 Room과 Player를 만들고, 더미클라이언트들이 서버에 로그인 - Room 입장까지 구현해봤다.

 

이제부터 JobQueue를 만들어보자.

---

JobQueue를 만드는 이유

지금 게임서버의 동작을 보면 문제가 몇가지 있다.

1. 네트워크 패킷이 들어올 때만 일을 하고, 그 외 게임로직은 실행되지 않음.
   유저패킷처리는 잘 해주고있는데 그 외 몬스터나 NPC의 움직임 등은 전혀 처리해주는 방법이 없다.
2. 네트워크 패킷을 받은 스레드가 로직도 실행하는데 이는 lock 경합으로 인한 병목현상을 발생시킨다.
   예를 들어 하나의 Room에서 10가지 일이 발생했는데, 이를 10개의 스레드가 받아서 처리를 시작했다 하면, Room에서의 대부분의 작업은 Lock을 걸고 처리하기 때문에 1개의 스레드만 일하고, 9개의 스레드는 놀게되는 문제가 생긴다. 9개의 스레드가 놀지않고 다른 일을 할 수 있도록 만들어줘야한다.

이러한 문제를 해결하기 위해 JobQueue를 만든다. 그리고 각각의 스레드는 그냥 lock을 걸고, JobQueue에 Job을 넣고 빠진 다음, 하나의 스레드만 Job에 들어간 일들을 처리하도록 만들면 된다. 하나의 스레드만 해도 되는게, 어차피 Lock을 걸고 처리할 일들이기 때문에 단일스레드로 돌려도 성능적으로 아무 차이가 없다.

 

게임서버 강의에서는 그 하나의 스레드는 가장 처음 JobQueue에 Job을 넣은 스레드로 설정했고, 해당 스레드가 일을 처리하는 동안에 다른 스레드들은 그냥 Job을 JobQueue에 집어넣고 다른 일을 하러 가도록 했다. Room안에서의 로직 처리처럼 "어차피 싱글스레드로 작동해야 할 일" 들을 하나의 JobQueue로 처리하는 느낌.

---

Job 클래스

우선 일을 담아놓고, 나중에 실행하기 위한 Job 클래스를 만들자.

#include <functional>

class Job
{
public:
	Job(function<void()>&& callback) : _callback(move(callback)) {}

	template<typename T, typename Ret, typename... Args>
	Job(shared_ptr<T> owner, Ret(T::* memFunc)(Args...), Args&&... args)
	{
		_callback = [owner, memFunc, args...]()
			{
				(owner.get()->*memFunc)(args...);
			}
	}

	void Execute()
	{
		_callback();
	}

private:
	function<void()> _callback;
};

---

LockQueue 클래스

큐인데 push, pop 할 때 Lock을 걸고 쓰기위한 큐.

JobQueue 내부에서 Job을 넣었다 뺐다 하기 위해 사용할 예정.

#pragma once

template<typename T>
class LockQueue
{
public:
	void Push(T item)
	{
		lock_guard<mutex> lock(_m);
		_items.push(item);
	}

	T Pop()
	{
		lock_guard<mutex> lock(_m);
		if (_items.empty())
			return T();

		T ret = _items.front();
		_items.pop();
		return ret;
	}

	T PopNoLock()
	{
		if (_items.empty())
			return T();

		T ret = _items.front();
		_items.pop();
		return ret;
	}

	void PopAll(OUT vector<T>& items)
	{
		lock_guard<mutex> lock(_m);
		while (T item = PopNoLock())
			items.push_back(item);
	}

	void Clear()
	{
		lock_guard<mutex> lock(_m);
		_items = queue<T>();
	}

private:
	mutex _m;
	queue<T> _items;
};

---

JobQueue 클래스

DoAsync함수로 함수를 Job으로 만들어 넣고, 내가(스레드) 첫번째로 넣은 스레드라면, Execute까지 하고, 아니면 push만 하고 빠져나온다. Execute를 하고 큐를 다시 확인하는데, 그새 새 Job이 큐에 차있다면 그것도 처리한다.

#pragma once
#include "Job.h"
#include "LockQueue.h"

class JobQueue : public enable_shared_from_this<JobQueue>
{
public:
	void DoAsync(function<void()>&& callback)
	{
		Push(make_shared<Job>(std::move(callback)));
	}

	template<typename T, typename Ret, typename... Args>
	void DoAsync(Ret(T::* memFunc)(Args...), Args... args)
	{
		shared_ptr<T> owner = static_pointer_cast<T>(shared_from_this());
		Push(make_shared<Job>(owner, memFunc, std::forward<Args>(args)...));
	}

	void					ClearJobs() { _jobs.Clear(); }

public:
	void					Push(JobRef job, bool pushOnly = false);
	void					Execute();

protected:
	LockQueue<JobRef>		_jobs;
	atomic<int32>			_jobCount = 0;
};

#include "pch.h"
#include "JobQueue.h"
#include "CoreTLS.h"

void JobQueue::Push(JobRef job, bool pushOnly)
{
	const int32 prevCount = _jobCount.fetch_add(1);
	_jobs.Push(job); // WRITE_LOCK

	// 첫번째 Job을 넣은 쓰레드가 실행까지 담당
	if (prevCount == 0)
	{
		// 이미 실행중인 JobQueue가 없으면 실행
		if (LCurrentJobQueue == nullptr && pushOnly == false)
		{
			Execute();
		}
	}
}

void JobQueue::Execute()
{
	LCurrentJobQueue = this;

	while (true)
	{
		vector<JobRef> jobs;
		_jobs.PopAll(OUT jobs);

		const int32 jobCount = static_cast<int32>(jobs.size());
		for (int32 i = 0; i < jobCount; i++)
			jobs[i]->Execute();

		// 남은 일감이 0개라면 종료
		if (_jobCount.fetch_sub(jobCount) == jobCount)
		{
			LCurrentJobQueue = nullptr;
			return;
		}
	}
}

 

LCurrentJobQueue는 스레드 로컬 변수를 쓰는 이유

결론부터 말하자면, 하나의 스레드가 하나의 JobQueue만 처리하도록 하기 위함이다.

 

LCurrentJobQueue를 사용하지 않으면 이런 일이 발생한다.

A, B 2개의 JobQueue가 있고, 10개의 스레드가 있다고 생각해보자.

5개의 스레드가 A에, 5개가 B 에 잡을 Push한다고 하자.

이 때 가장 먼저 실행된 1번스레드가 A에 잡을 넣고 실행까지 하는데 그 잡의 내용이 B 잡큐에 잡을 Push하는 내용이라면, 1번 스레드가 A와 B JobQueue를 모두 붙잡고있고, 다른 9개의 스레드는 A,B에 잡을 넣기만 한다. 원래 계획은 각 잡큐의 일처리는 하나의 스레드가 맡는건데, 2개의 잡큐를 하나의 스레드가 맡아버리는 것. 2개라 다행이지 100개의 잡큐를 1개의 스레드가 맡아버린다 생각하면 제대로 돌아갈 수가 없을것이다.

 

그래서 JobQueue::Execute를 호출할 때 스레드 로컬 변수 LCurrentJobQueue에 JobQueue를 등록함으로써 "나는 지금 A 잡큐 담당 스레드니까 다른 잡큐에는 잡을 Push만 하고 실행은 하지 않겠음" 을 적어두는 것이다. (굳이 JobQueue 포인터로 할 필요 없이 bool로 해도 되지 않을까 싶긴 한데 내가 담당하는 JobQueue라면 execute까지 하도록 설계를 변경한다면 필요할지도?)

---

Room 클래스 변경

이제 Room은 JobQueue를 상속받는다.

이제부터 각 Room에서 호출하는 함수는 JobQueue의 DoAsync함수로 "jobqueue에 등록 -> 싱글스레드로 동작" 한다. 때문에 기존에 뮤텍스를 잡던걸 다 없애줘도 된다. Room에 대해 1명이 일하는데 9명이 대기하는 상황이 없어진 것.

 

class Room : public JobQueue
{
public:
	Room();
	~Room();

	void Enter(PlayerRef player);
	void Leave(PlayerRef player);
	void Broadcast(SendBufferRef sendBuffer);
private:
	uint32 _roomId;
	map<uint64, PlayerRef> _players;
};

extern RoomRef GRoom[2];

RoomRef GRoom[2];

void Room::Enter(PlayerRef player)
{
	_players.insert({ player->_info.id(), player });

	Protocol::S_CHAT pkt;
	player->_room = static_pointer_cast<Room>(shared_from_this());
	string msg = player->_info.name() + " Entered Room" + to_string(_roomId);
	pkt.set_msg(msg);

	Broadcast(ClientPacketHandler::MakeSendBuffer(pkt));
}

void Room::Leave(PlayerRef player)
{
	_players.erase(player->_info.id());
	
	Protocol::S_CHAT pkt;
	string msg = "Room " + to_string(_roomId) + " : " + player->_info.name() + " Left Room.";
	pkt.set_msg(msg);
	Broadcast(ClientPacketHandler::MakeSendBuffer(pkt));
}

void Room::Broadcast(SendBufferRef sendBuffer)
{
	for (auto& player : _players)
	{
		if (auto owner = player.second->_owner.lock())
		{
			owner->Send(sendBuffer);
		}
	}
}

Room의 멤버함수를 DoAsync로 사용하려면 자기 자신을 shared_from_this로 만들 수 있어야 하니까 일단 extern Room GRoom을 RoomRef GRoom으로 바꿔주자. 그리고 GameServer main문에서 생성해주자. 일단은 임시다.

#include "Room.h"

int main()
{
	cout << "=== Server ===" << endl;

	//Room 테스트용
	GRoom[0] = make_shared<Room>();
	GRoom[1] = make_shared<Room>();
    /*...*/
}

 

Player 클래스 변경

원래 owner를 shared_ptr로 들고있었는데, weak_ptr로 변경했다. 순환참조 문제때문.

그리고 Room과 ChatTest도 추가했다. 클라이언트로부터 C_CHAT 패킷이 오면 패킷핸들러에서 Player::ChatTest를 호출해주자.

#pragma once
#include "Protocol.pb.h"

class Room;

class Player
{
public:
	Player(const Protocol::Player& player, GameSessionRef owner);
	~Player();

	void ChatTest(const string& msg);

	Protocol::Player _info;
	weak_ptr<GameSession> _owner;
	weak_ptr<Room> _room;
};

void Player::ChatTest(const string& msg)
{
	if (RoomRef room = _room.lock())
	{
		Protocol::S_CHAT pkt;
		string chatMsg = _info.name() + " : " + msg;
		pkt.set_msg(chatMsg);

		room->DoAsync(&Room::Broadcast, ClientPacketHandler::MakeSendBuffer(pkt));
	}
}

 

---

테스트

테스트를 위해 Room의 멤버함수를 직접 사용하던 부분을 모두 DoAsync를 사용하도록 변경해야한다.

 

Handle_C_ENTER_ROOM

 

void Handle_C_ENTER_ROOM(const PacketSessionRef& session, const Protocol::C_ENTER_ROOM& pkt)
{
	/*...*/
    PlayerRef player = make_shared<Player>(playerInfo, gameSession);
    gameSession->_player = player;
	RoomRef room = GRoom[pkt.room_id()];
	room->DoAsync(&Room::Enter, player);
	response.set_success(true);

	session->Send(ClientPacketHandler::MakeSendBuffer(response));
}

 

Handle_C_CHAT

void	Handle_C_CHAT(const PacketSessionRef& session, const Protocol::C_CHAT& pkt)
{
	GameSessionRef gameSession = static_pointer_cast<GameSession>(session);
	if (gameSession->_player == nullptr)
		return;
	gameSession->_player->ChatTest(pkt.msg());
}

 

그리고 더미클라이언트에서도 0.2초마다 C_CHAT 패킷을 보내도록 만들어봤다.

int main()
{
	cout << "=== Client ===" << endl;
	/*...*/
    
	this_thread::sleep_for(chrono::seconds(1));
	while (true)
	{
		this_thread::sleep_for(chrono::milliseconds(200));
		Protocol::C_CHAT pkt;
		pkt.set_msg("Hello Server!");
		service->broad_cast_test(ServerPacketHandler::MakeSendBuffer(pkt));
	}
}

 

TLS 코드에 버그가 또 있어서 해결한다음 테스트해봤다. (버그해결은 밑에)

잘 된다.

---

TLS 버그 해결

사소한 문제와 큰 문제가 있었는데 결국 해결했다.

JobQueue의 DoAsync를 적용하기 전에 더미클라에서 C_CHAT을 보내도록 바꾸고 먼저 돌려봤는데 이상하게 돌아가다가 클라이언트가 하나씩 disconnect되는 문제가 발생했다. 그래서 TLS세션을 떼고 암/복호화 없이 다시 실행시켰는데, 이번엔 아예 패킷 송수신이 제대로 안되는 문제가 생겼다.

 

평문통신에서 패킷 송수신 안되는 문제 (사소한 문제)

//virtual uint8 Decrypt(RecvBuffer& encBuffer, RecvBuffer& decBuffer) { return 0; }
virtual uint8 Decrypt(RecvBuffer& encBuffer, RecvBuffer& decBuffer) { return 1; }

이전에 ProcessRecv에서 Decrypt를 반복적으로 하는 코드를 바꿨었는데, 평문통신의 Decrypt의 리턴값을 바꾸지 않아서 생기는 문제였다. 바꿔주니 평문통신에서는 잘 돌아갔고, 하나씩 disconnect 되는 문제가 발생하지 않았다.

 

 

TLS 통신에서 disconnect 되는 문제 (큰 문제)

우선 Disconnect되면 안되는 상황이긴 하더라도 Disconnect가 정상적으로 이루어졌기 때문에 도대체 어디서 호출되는지 RegisterDisconnect에 Crash를 넣고, 호출스택을 봤다. ProcessRecv의 Decrypt함수가 심각한 SSL 에러를 리턴해서 RegisterDisconnect를 호출하고있었다. 복호화 과정에서 SSL_read_ex함수가 SSL_ERROR_SYSCALL 리턴한 것.

 

recv과정에서는 Rbio, SSL_read만 사용하고, send과정에서는 Wbio, SSL_write만 사용하기 때문에 적어도 send와 recv는 완전히 분리되어있다고 생각하고, recv 과정은 반드시 싱글스레드로 동작하는데 왜 여기서 SSL_ERROR_SYSCALL 에러가 발생하는지 이해할 수가 없었다. 다만 send과정은 동시에 일어날 수 있기 때문에 wbio에 암호화 하고, 꺼내고 할 때 문제가 있을 것 같기는 했다. 근데 그러면 send과정에서 문제가 터져야지 왜 recv 과정에서 문제가 터지는지도 이해가 안됐다.

 

AI에게 물어보니 ssl에 rbio와 wbio가 분리해서 사용하더라도 ssl 내부적으로 암 복호화에 record layer는 하나이기 때문에 하나의 ssl 객체에 접근하는 스레드는 반드시 하나여야한다고 한다.

 

그래서 우선 Encrypt와 Decrypt를 하는 부분에서 lock_guard을 걸었더니 해결이 됐다.

해결은 됐는데 아무래도 송 수신을 할때마다 lock을 걸고 하는것이다 보니 송 수신이 많아질수록 스레드가 대기하게되는 심각한 문제가 있을 것 같다. 일단은 이렇게 처리하고 넘어가지만, 만약 제대로 해결이 안된다면 아마 TLS를 버려야하지 않을까 싶다.

Send

void	 Session::Send(SendBufferRef sendBuffer)
{
	SendBufferRef encBuffer;
	{
		lock_guard<mutex> lock(_m);
		bool isSuccess = Encrypt(sendBuffer, encBuffer);
		if (isSuccess == false)
			return;
	}

	{
		lock_guard<mutex> lock(_m);
		_sendBuffers.push(encBuffer);
	}

	bool expected = false;
	if (_sendRegistered.compare_exchange_strong(expected, true))
	{
		RegisterSend();
	}
}

 

ProcessRecv

void Session::ProcessRecv(int32 numOfBytes)
{
	_recvEvent.Clear();
	if (numOfBytes == 0) // 클라이언트가 정상적으로 연결을 종료한 경우
	{
		RegisterDisconnect();
		return;
	}

	RecvBuffer& encBuffer = GetEncRecvBuffer();
	RecvBuffer& decBuffer = GetDecRecvBuffer();

	encBuffer.OnWrite(numOfBytes);

	{
		lock_guard<mutex> lock(_m);
		// enc의 데이터를 복호화 해서 dec로 이동
		// encBuffer.OnRead, decBuffer.OnWrite는 내부에서 호출해줌
		bool repeat = true;
		while (repeat)
		{
			uint8 ret = Decrypt(encBuffer, decBuffer);
			switch (ret)
			{
			case 0: // 성공. 복호화 할 데이터 더 있을 수 있음. 반복해서 복호화 시도.
				break;
			case 1: // 복호화 데이터 부족
				repeat = false;
				break;
			case 2: // 상대가 shutdown. shutdown 호출 가능
				//TODO shutdown 정상종료
				repeat = false;
				break;
			case 3: // 에러. shutdown 호출 불가능.
				RegisterDisconnect();
				repeat = false;
				break;
			}
		}
	}
	/*...*/
}

---

현재까지의 git 버전

Feat: JobQueue 추가 · Dodontak/Project_Island_GameServer@2d07b11