[C++ 게임 서버] 6-6. JobQueue #5

Rookiss 님의 '[C++과 언리얼로 만드는 MMORPG 게임 개발 시리즈] Part4: 게임 서버' 를 들으며 배운 내용을 정리하고 있습니다. 관심이 있으신 분은 꼭 한 번 들어보시기를 추천합니다!

[C++ 게임 서버] 6-6. JobQueue #5

핵심 :

1. JobQueue 방식을 이용해 쓰레드가 Job 을 처리할 때 발생하는 병목 현상 문제를 해결해 보자.

2. GlobalQueue 와 Tick 의 개념을 활용해, 이전 글에서 언급했던 두 가지 병목 현상을 해결할 수 있다.

3. 서버 로직과 클라 로직은 각각 분리하여 돌아가도록 구현하는 것이 좋다. 다만 이번 글에서는, DoWorkerJob 함수에서 모든 로직을 한꺼번에 처리하도록 구현하였다.

---

이번 시간에는 저번 글에서 지적했던 병목 현상을 해결하기 위해, GlobalQueue 와 시간 제한을 주는 방식을 도입해 보자.

일단 Job 부분은 달라진 것이 없는데... JobQueue 에서 Execute 를 해주는 로직을 일부 수정할 것이다. 거두절미하고, 바로 JobQueue 의 구현을 보자. 일단 Push 부터 시작하겠다.

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

	// 첫번째 Job을 넣은 쓰레드가 실행까지 담당
	if (prevCount == 0)
	{
		// 이미 실행중인 JobQueue가 없으면 실행
		if (LCurrentJobQueue == nullptr)
		{
			Execute();
		}
		else
		{
			// 여유 있는 다른 쓰레드가 실행하도록 GlobalQueue에 넘긴다
			GGlobalQueue->Push(shared_from_this());
		}
	}
}

위의 Push 함수에서는 prevCount 를 체크하기는 하지만, LCurrentJobQueue 라는 것을 추가로 체크하고, nullptr 가 아니라면 GGlobalQueue 에 해당 JobQueue 를 넘기고 있다.

일단 LCurrentJobQueue 가 무엇일까? 🤔

thread_local JobQueue*			LCurrentJobQueue = nullptr;

 

우리는 CoreTLS.cpp 에 LCurrentJobQueue 를 위와 같이 JobQueue* 타입으로 선언해 줄 것이다. 즉, 해당 쓰레드가 물고 있는 JobQueue 를 나타낸다.

 

그리고 GGlobalQueue 는... CoreGlobal 에 다음과 같이 선언되어 있다.

// 나머지는 전부 생략
GlobalQueue*		GGlobalQueue = nullptr;

class CoreGlobal
{
public:
	CoreGlobal()
	{
		GGlobalQueue = new GlobalQueue();
	}

	~CoreGlobal()
	{
		delete GGlobalQueue;
	}
} GCoreGlobal;

GlobalQueue 클래스는 다음과 같이 구현되어 있다 :

class GlobalQueue
{
public:
	GlobalQueue() {};
	~GlobalQueue() {};

	void					Push(JobQueueRef jobQueue)
	{
		_jobQueues.Push(jobQueue);
	}
	JobQueueRef				Pop()
	{
		return _jobQueues.Pop();
	}

private:
	LockQueue<JobQueueRef> _jobQueues;
};

단순히 LockQueue 타입의 JobQueue 를 들고 있는 클래스이다. 다만 '전역'이라는 것이 특이한 점이다.

 

즉, JobQueue::Push 의 바뀐 로직을 설명하자면... 

특정 쓰레드에 대해, 해당 쓰레드가 JobQueue 에 첫번째 Job 을 넣었으면서, 동시에 이 쓰레드가 다른 JobQueue 를 Execute 를 하고 있지 않은 경우에만 실제로 일감을 처리하고, 그렇지 않으면 전역으로 만들어진 JobQueue 에 대한 Queue(GGlobalQueue)에 해당 JobQueue 를 통째로 넣어준다.

이어서 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;
		}

		const uint64 now = ::GetTickCount64();
		if (now >= LEndTickCount)
		{
			LCurrentJobQueue = nullptr;
			// 여유 있는 다른 쓰레드가 실행하도록 GlobalQueue에 넘긴다
			GGlobalQueue->Push(shared_from_this());
			break;
		}			
	}
}

Execute 에서, LCurrentJobQueue 에 this 를 넣어 주는 것을 확인할 수 있다. 일감을 처리하는 로직은 거의 흡사하다.

다만 추가적으로, Tick 을 체크해서 now 가 일정 틱 수치보다 크면 일감을 처리하지 않고 JobQueue 에 있는 남은 Job 들에 대한 처리를 GGlobalQueue 에 넣어 다른 쓰레드가 처리하도록 만들고 있다. 이는, 특정 쓰레드가 너무 과도하게 많은 시간을 Execute 에 쏟지 않도록 간단하게 만든 제약이라고 볼 수 있다. 😉

 

그리고 이제 Job 을 처리하는 것도 쓰레드에게 직접 위임할 수 있으므로, ThreadManager 에서 다음 함수를 하나 추가해 주자.

// ThreadManager.h
static void DoGlobalQueueWork();

// ThreadManager.cpp
void ThreadManager::DoGlobalQueueWork()
{
	while (true)
	{
		uint64 now = ::GetTickCount64();
		if (now > LEndTickCount)
			break;

		JobQueueRef jobQueue = GGlobalQueue->Pop();
		if (jobQueue == nullptr)
			break;

		jobQueue->Execute();
	}
}

DoGlobalQueueWork 함수는 틱을 체크하면서 GlobalQueue 에서 JobQueue 를 꺼내 Execute 를 호출해서 일감을 처리하고 있다!

 

그럼 이제 GameServer 에서는, 일감을 처리하는 함수를 아래와 같이 정의할 수 있게 된다 :

enum
{
	WORKER_TICK = 64
};

void DoWorkerJob(ServerServiceRef& service)
{
	while (true)
	{
		LEndTickCount = ::GetTickCount64() + WORKER_TICK;

		// 네트워크 입출력 처리 -> 인게임 로직까지 (패킷 핸들러에 의해)
		service->GetIocpCore()->Dispatch(10);

		// 글로벌 큐
		ThreadManager::DoGlobalQueueWork();
	}
}

즉, 각 쓰레드가 JobQueue 를 처리하려고 할때, WORKER_TICK 이상의 시간이 걸린다면, JobQueue::Execute 함수를 돌이켜 볼때, 빠져나오게 된다는 것을 유추할 수 있다. 물론 이 값은 지금은 임의의 고정 값으로 설정했지만, 실제 프로젝트에서는 상황에 맞게 조금씩 변동하도록 만들긴 할 것이다. 🤣

일단 핵심은, 이제 DoWorkerJob 내에서 Dispatch 를 하게 되는데, 이때 인자 10을 줘서 Timeout 시간을 10ms 로 두었다. 즉, 일정 시간이 지나면 무한으로 대기하지 말고 빠져 나오라는 얘기이다.

마지막으로, 우리가 만든 DoWorkerJob 은 사실... '만능형 일꾼' 함수이다. 즉, 게임을 실제로 만들면, 어떤 로직은 서버에서 돌아가야 하지만, 어떤 로직은 클라이언트에서만 돌아도 된다(예를 들어, 화염을 던졌을 때 이펙트 렌더링 같은 것). 그런데 일단은, 우리는 모든 로직이 서버에서 돌아간다고 생각하고, ThreadManager::DoGlobalQueueWork(); 를 DoWorkerJob 안에 넣어줄 것이다.

만약 서버 상에서의 로직과 클라이언트 상의 로직이 분리되어 돌아가야 한다면, ThreadManager::DoGlobalQueueWork(); 부분을 밖으로 빼야 할 것이다. 😊

 

어쨌든, 이제 GameServer 의 main 함수는 아래와 같게 수정된다 :

int main()
{
	ClientPacketHandler::Init();

	ServerServiceRef service = MakeShared<ServerService>(
		NetAddress(L"127.0.0.1", 7777),
		MakeShared<IocpCore>(),
		MakeShared<GameSession>, // TODO : SessionManager 등
		100);

	ASSERT_CRASH(service->Start());

	for (int32 i = 0; i < 5; i++)
	{
		GThreadManager->Launch([&service]()
			{
				DoWorkerJob(service);
			});
	}

	// Main Thread
	DoWorkerJob(service);

	GThreadManager->Join();
}

이제 각 쓰레드는 Job 을 Push 도 하고, Execute 도 하고, 뭐 다 할 것이다!