Unreal에서 Multithreading으로 Observer Pattern 최적화하기

프로젝트 : https://github.com/HoonInPark/Hyperion.git
본 포스트에 대한 내용은 feat/obop 브랜치에 있다.

사담

누가 말했는데, 고수 개발자들이 쓴 네이버 블로그의 낭만은 개발이랑 상관 없는 소리도 같이 섞여 있는 거라고.
내가 고수도 아니고 여긴 네이버 블로그도 아니지만 개발이랑 상관 없이 개발할 당시 내 기분도 간략히 써보려 한다.

최근 이직 준비로 개발을 많이 못했다.
마냥 회사 연락 기다리기엔 불안 심심해서 다시 하고 싶었던 것들을 해보려 한다.
역시 개발자의 정체성은 만드는 것, '만든다 고로 나는 존재한다'
거의 2주만인가, 컴파일러 돌면서 CPU 온도 높아지는 걸 오랜만에 보니 즐겁다.

아, 그리고 이게 벨로그 첫 글인데, 아는 거 정리하는 거 말고, 내가 모르는 것(그리고 남들도 모르는 것)을 연구하는 곳으로 쓰고자 한다.
남들 다 아는 배열, 리스트 이런 거 정리해봤자 면접관 말고는 아무도 안본다.

문제

먼저, Hyperion이라는 언리얼 프로젝트에서 옵저버 패턴을 구현하여 클라-서버 연동을 구현했다. 언리얼에서 옵저버 패턴을 구현하는 방법은 다음 포스트를 참고하라.
https://velog.io/@apzl79/Unreal-Engine-5-22-Observer-Pattern

나의 경우 다음과 같이 구성했다.
우선 언리얼 UObject를 상속한 클래스에 적용할 수 있는 옵저버 인터페이스다.

// Fill out your copyright notice in the Description page of Project Settings.

#pragma once

#include "CoreMinimal.h"
#include "UObject/Interface.h"
#include "HyperionBase/ObservableBase.h"

#include "ObserverBase.generated.h"

// This class does not need to be modified.
UINTERFACE(MinimalAPI)
class UObserverBase : public UInterface
{
	GENERATED_BODY()
};

/**
* 
*/
class HYPERIONBASE_API IObserverBase
{
	GENERATED_BODY()

	// Add interface functions to this class. This is the class that will be inherited to implement this interface.
public:
	UFUNCTION(BlueprintCallable, BlueprintNativeEvent, Category = "Observer")
	void OnNotify(UObservableBase* _pInObservable);
	
};

그리고 관측 대상, 옵저버블(Observable) 클래스는 UActorComponent를 상속 받았다.

// Fill out your copyright notice in the Description page of Project Settings.

#pragma once

#include "CoreMinimal.h"
#include "Components/ActorComponent.h"
#include "ObservableBase.generated.h"

class IObserverBase;

UCLASS( ClassGroup=(Custom), meta=(BlueprintSpawnableComponent) )
class HYPERIONBASE_API UObservableBase : public UActorComponent
{
	GENERATED_BODY()

public:
	// Sets default values for this component's properties
	UObservableBase();

public:
	virtual void Subscribe(TScriptInterface<IObserverBase> _Observer);
	virtual void Unsubscribe(TScriptInterface<IObserverBase> _Observer);

	void NotifyObservers();

private:
	TArray<TScriptInterface<IObserverBase>> Observers;
};

그리고 구현부...

// Fill out your copyright notice in the Description page of Project Settings.


#include "ObservableBase.h"

#include "ObserverBase.h"

// Sets default values for this component's properties
UObservableBase::UObservableBase()
{
	// Set this component to be initialized when the game starts, and to be ticked every frame.  You can turn these features
	// off to improve performance if you don't need them.
	PrimaryComponentTick.bCanEverTick = false;

	// ...
}

void UObservableBase::Subscribe(TScriptInterface<IObserverBase> _Observer)
{
	Observers.Add(_Observer);
}

void UObservableBase::Unsubscribe(TScriptInterface<IObserverBase> _Observer)
{
	Observers.Remove(_Observer);
}

void UObservableBase::NotifyObservers()
{
	for (TScriptInterface<IObserverBase> Observer : Observers)
	{
		if (Observer.GetObject() && Observer.GetObject()->GetClass()->ImplementsInterface(UObserverBase::StaticClass()))
		{
			IObserverBase::Execute_OnNotify(Observer.GetObject(), this);
		}
	}
}

나의 프로젝트에서 보통 관측 대상은 액터 클래스이고 관측자, 옵저버는 어떤 클래스든 될 수 있기에 위와 같이 만들었다.

여기서 옵저버 패턴에 흔히 제기되는 문제, NotifyObservers()로 OnNotify(...) 콜백 호출이 동기적이라는 문제가 있다.
특히나 나의 경우 서버에 접속 중인 여러 클라이언트의 캐릭터들이 옵저버 패턴으로 Replicate되고 있기에 동접자가 많아지면 그만큼 저 옵저버 배열이 커지게 되고, for문도 오래 걸린다.
최악의 경우 100명 넘는 유저가 한 화면에 담기는 리니지 공성전이 있겠다.

이 경우 거리가 멀거나 눈에 안보이면 동기화를 잠시 중단하는 등의 최적화 방법은 쓸 수 없다.

또, 저기서 호출하는 콜백 중 하나라도 좀 긴 작업이면 큰일이다.

해결

당장 떠오르는 건 Mutithreading이다.
이젠 좀 느리다 싶으면 이 방법부터 생각해 보는 거 같다.

언리얼에서 제공하는 FQueuedThreadPool을 사용해보면 어떨까 했지만,
이는 동접자가 많지 않을 때도 pool 내 모든 가용 thread가 작업을 하고 있는 상황이 생길 것 같았다.
이게 왜 문제가 되냐?
30개의 작업이 큐잉돼 있는 상황이라 생각해보면,
thread가 10개 담긴 FQueuedThreadPool를 사용하면 모든 스레드에게 3개씩 작업이 분배된다.
CPU 입장에선 10개의 코어 각각이 컨텍스트 스위칭을 겪는 것이다.

내가 생각해낸 방법은 다음과 같다.

CPU 코어 갯수만큼 FRunnable 클래스를 미리 생성해 두고...

int32 NumLogicalCores = FPlatformMisc::NumberOfCoresIncludingHyperthreads();
for (int32 i = 0; i < NumLogicalCores - 3; i++)
{
	m_ReplicationRunnables.Add(new FReplicationRunnable);
}

배열로 관리한다.

TArray<FReplicationRunnable*> m_ReplicationRunnables;

각 FRunnable 클래스는 멤버로 크기가 제한된 작업 큐를 가진다.

TCircularQueue<TFunction<void()>> m_TaskQ;

각 FRunnable 객체 내 thread에선 열심히 Dequeue()하고 있다.

만약 첫번째 FRunnable 객체의 작업 큐가 다 차 있다면, 두번째 FRunnable 객체의 작업큐로 가서 Enqueue()한다.
반대로 각 FRunnable 객체는 작업 큐가 비어 있을 경우 Wait()가 호출된 상태다.
즉, block 되어 CPU는 여기에 실행시간을 할애하지 않는다.

결과적으로 보면, 동접자 수에 따라 동작하는 thread의 갯수가 달라진다.
다만 작업 큐 크기를 어느 정도로 할지는 여러 하드웨어에서 테스트해보고 결정할 일이다.
최신 유행하는 멀티플레이 게임의 권장 사양을 기준으로 정책을 세우면 좋을 듯 하다.

일단 완성된 FRunnable 클래스.

class HYPERION_API FReplicationRunnable : FRunnable
{
public:
	FReplicationRunnable();
	virtual ~FReplicationRunnable() override;

	virtual bool Init() override;
	virtual uint32 Run() override;
	virtual void Stop() override;
	//virtual void Exit() override;

	FORCEINLINE bool EnqueueTask(TFunction<void()> _InTask)
	{ 
		return m_TaskQ.Enqueue(_InTask); 
	}

private:
	FRunnableThread* m_pThread{ nullptr };
	TCircularQueue<TFunction<void()>> m_TaskQ;
};