JobQueue(5)

 

CoreTLS 코드 수정

//추가
extern thread_local class JobQueue* LCurrentJobQueue;
extern thread_local uint64 LEndTickCount;

 

• LCurrentJobQueue : 내가 실행하고 있는 JobQueue를 확인하기 위한 JobQueue 포인터 추가
• LEndTickCount : 틱이 완료되는 시간

 

---

 

전에 말했듯이 하나의 스레드에 일감이 몰리게 될 수도 있다.

이를 해결하는 가장 간단한 방법은 스레드가 JobQueue를 호출하고 있는지 아닌지를 판별하여 

이미 호출하고 있다면 처음으로 job을 넣었다고 하더라도 실행시키지 않고 다른 스레드로 넘기게 될 것이다.

 

JobQueue.cpp 수정

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

	//맨 처음으로 일감을 넣은 상태
	//(첫번째 Job을 넣은 쓰레드가 실행까지 담당
	if(prevCount == 0)
	{
		if (LCurrentJobQueue == nullptr)
		{
			Execute();
		}
		else
		{
			GGlobalQueue->Push(shared_from_this());
		}
	}
}

• if (LCurrentJobQueue == nullptr)
  이미 실행중인 JobQueue가 없으면 실행
  
  
• else
  여유 있는 다른 쓰레드가 실행하도록 GlobalQueue에 넘긴다.
  
  

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

	while (true)
	{
		xvector<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;
		}
	}
}

 

---

 

GlobalQueue에 넣은 일감을 실행시켜주는 코드를 ThreadManager에 추가.

 

ThreadManager.cpp

void ThreadManager::DoGlobalQueueWork()
{
	while (true)
	{
		uint64 now = ::GetTickCount64();

		if (now > LEndTickCount)
			break;

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

		jobQueue->Execute();
	}
}

 

---

 

기존에 스레드를 사용하는 방식으로는

1. 워커 쓰레드가 반복문을 통해 IocpCore를 확인하여 Dispatch 함수를 호출.

2. Dispatch 함수 아네서는 Iocp를 GetQueuedCompletionStatus 함수를 통해 입출력이 완료되기를 대기.

3. recv나 send 같은 네트워크 이벤트 통지가 완료되면 빠져나와서  세션 또는 리스너의 Dispatch를 호출.

4. (세션의 경우) Dispatch를 통해 페킷 세션에서 패킷을 조립해서 컨텐츠 코드로 넘어옴.

 

네트워크 입출력 처리를 하면서 인게임 로직까지 같이 처리해주고 있는 방식이다.

 

그런데 기존 Dispatch 함수는 사용할 때 타임 아웃 시간을 따로 저애주지 않고 사용하고 있어서

네트워크 입출력이 완료 될 때 까지 함수가 무한대로 대기하는 상황이 발생할 수 있다.

 

이런 문제점을 개선함과 동시에 최종적으로 우리가 스레드를 사용할 방식은

네트워크 이벤트, 인게임 로직, 글로벌 큐를 비우는 동작 전부 수행하는 방식을 사용할 것이다.

 

void DoWorkerJob(ServerServiceRef& service)
{
	while (true)
	{
		LEndTickCount = ::GetTickCount64() + WORKER_TICK;
		service->GetIocpCore()->Dispatch(10);

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

 

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]()
			{
				while (true)
				{
					//service->GetIocpCore()->Dispatch();
					DoWorkerJob(service);
				}
			});
	}

	//메인 스레드
	DoWorkerJob(service);

 	GThreadManager->Join();
}

 

 

메인 스레드와 워커스레드의 역할이 모두 같기 때문에( 네트워크 이벤트, 인게임 로직, 글로벌 큐 처리) 똑같은 DoWorkerJob() 함수를 호출한다.