언리얼 엔진 스레드 (Recv)

전에 껍데기만 만들어두었던 PacketSession과 NetworkWorker 클래스를 작업할 차례다.

---

NetworkWorker.h

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

#pragma once

#include "CoreMinimal.h"
#include "Containers/Queue.h"

class FSocket;

struct IOCPUNREAL_API FPacketHeader
{
	FPacketHeader() : PacketSize(0), PacketID(0)
	{
	}

	FPacketHeader(uint16 PacketSize, uint16 PacketID) : PacketSize(PacketSize), PacketID(PacketID)
	{
	}
	
	friend FArchive& operator<<(FArchive& Ar, FPacketHeader& Header)
	{
		Ar << Header.PacketSize;
		Ar << Header.PacketID;
		return Ar;
	}

	uint16 PacketSize;	//2바이트
	uint16 PacketID;	//2바이트
};

class IOCPUNREAL_API RecvWorker : public FRunnable
{
public:
	RecvWorker(FSocket* Socket, TSharedPtr<class PacketSession> Session);
	~RecvWorker();

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

	void Destroy();

private:
	bool ReceivePacket(TArray<uint8>& OutPacket);
	bool ReceiveDesiredBytes(uint8* Results, int32 Size);


protected:
	FRunnableThread* Thread = nullptr;
	bool Running = true;
	FSocket* Socket;
	TWeakPtr<class PacketSession> SessionRef;
};

 

 

#include "Network/NetworkWorker.h"
#include "Sockets.h"
#include "Serialization/ArrayWriter.h"
#include "PacketSession.h"

RecvWorker::RecvWorker(FSocket* Socket, TSharedPtr<class PacketSession> Session) :Socket(Socket), SessionRef(Session)
{
	Thread = FRunnableThread::Create(this, (TEXT("RecvWorkerThread")));
}

RecvWorker::~RecvWorker()
{
}

bool RecvWorker::Init()
{
	GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Red, FString::Printf(TEXT("Recv Thread Init")));
	return true;
}

uint32 RecvWorker::Run()
{
	while (Running)
	{
		TArray<uint8> Packet;

		if (ReceivePacket(OUT Packet))
		{
			if (TSharedPtr<PacketSession> Session = SessionRef.Pin())
			{
				Session->RecvPacketQueue.Enqueue(Packet);
			}
		}
	}

	return 0;
}

void RecvWorker::Exit()
{
}

void RecvWorker::Destroy()
{
	Running = false;
}

bool RecvWorker::ReceivePacket(TArray<uint8>& OutPacket)
{
	const int32 HeaderSize = sizeof(FPacketHeader);
	TArray<uint8> HeaderBuffer;
	HeaderBuffer.AddZeroed(HeaderSize);

	if (ReceiveDesiredBytes(HeaderBuffer.GetData(), HeaderSize) == false)
		return false;

	//패킷 ID, Size 추출
	FPacketHeader Header;
	{
		FMemoryReader Reader(HeaderBuffer);
		Reader << Header;
		UE_LOG(LogTemp, Log, TEXT("Recv PacketID : %d, PacketSize : %d"), Header.PacketID, Header.PacketSize);
	}

	//패킷 헤더 복사
	OutPacket = HeaderBuffer;

	//패킷 내용 파싱
	TArray<uint8> PayloadBuffer;
	const int32 PayloadSize = Header.PacketSize - HeaderSize;
	OutPacket.AddZeroed(PayloadSize);

	if (ReceiveDesiredBytes(&OutPacket[HeaderSize], PayloadSize))
		return true;

	return false;
}

bool RecvWorker::ReceiveDesiredBytes(uint8* Results, int32 Size)
{
	//접속이 종료되었을 때 패킷 리시브 사이즈가 0으로 오기 때문에 이걸로 판단하면 된다.
	uint32 PendingDataSize;
	if (Socket->HasPendingData(PendingDataSize) == false || PendingDataSize <= 0)
		return false;

	int32 Offset = 0;

	while (Size>0)
	{
		int32 NumRead = 0;
		Socket->Recv(Results + Offset, Size, OUT NumRead);
		check(NumRead <= Size);

		if (NumRead <= 0)
			return false;

		Offset += NumRead;

		//다 읽었으니 Size를 0으로 만들고 while문 빠져나오기
		Size -= NumRead; 
	}

	return true;
}

 

 

TCP는 패킷의 경계선이 없다.
그래서 완전한 패킷으로 왔는지 판별할 수 있어야 한다.

가장 좋은 방법은 패킷 헤더를 두는 것.

 

struct IOCPUNREAL_API FPacketHeader

{

...

friend FArchive& operator<<(FArchive& Ar, FPacketHeader& Header)

...

}

• 언리얼에서 제공하는 기본적인 메모리 직렬화 객체
• 언리얼에서는 메모리 읽기, 쓰기 기능을 지원하기 위해 
• FMemoryArchive를 상속하여 FMemoryReader, FMemoryWriter 형태로 제공한다

 

send를 담당하는 worker와 recv를 담당하는 worker를 따로 만들어 줄 것이다.

지금은 우선 간단한 RecvWorker만 만들었다.

언리얼에서 스레드를 사용하기위해서는 FRunnable 클래스를 상속받고 3개의 함수를 오버라이딩 해주면된다.

 

 

bool RecvWorker::ReceivePacket(TArray<uint8>& OutPacket)

• ReceiveDesiredBytes 함수를 여기서 호출.
• 처음에 패킷 헤더를 파싱해서 ReceiveDesiredBytes 함수를 호출하여 패킷 헤더 데이터를 받아 
  패킷의 ID와 Size를 추출한다.
• 그 후, 패킷 헤더를 복사 한다음, 패킷에서 패킷헤더의 크기를 제외한 패킷 데이터크기를 저장한 버퍼를 매개변수로
  다시 ReceiveDesiredBytes 함수를 호출하여 데이터를 수신한다.

 

 

bool RecvWorker::ReceiveDesiredBytes(uint8* Results, int32 Size)

• 원하는 크기의 패킷이 올때까지 기다렸다가 Recv를 호출해주고 성공 여부결과를 리턴해주는 함수.

 

이제 위의 과정을 거쳐서패킷이 완성되고 조립이 끝났다.

그럼 이제 조립된 패킷을 통해 패킷을 처리해주는 코드를 호출해주어야 한다.

기존에 이 부분을 PacketHandler 클래스를 통해 자동화 해두었다.

하지만 지금 바로 전과 같은 방법으로 처리하려고 시도하면 크래시가 발생할 것이다.

 

왜냐하면 유니티와 마찬가지로 엔진에서 별도로 생성한 스레드에서 메인 스레드에 바로 접근하려고 하면 크래시가 발생한다.

이를 해결하기 위해 조립 해놓은 패킷을 Queue에 보관했다가 메인스레드에서 꺼내 쓰도록 분리해줘야 한다.

이 Queue를 어디에 보관하는지는 개인의 선택이지만 여기서는 PacketSession에 넣어두었다.

 

weak_ptr를 사용하면서 귀찮은 부분이 하나 있다.

위에 코드에서 보이듯이 shared_ptr을 weak_ptr로 변환해주어야 한다.

기존 C++ 표준에서는 lock()함수를 사용했지만 언리얼에서는 Pin()함수를 사용해주면 된다.

---

PacketSession.h

#pragma once

#include "CoreMinimal.h"

class IOCPUNREAL_API PacketSession :public TSharedFromThis<PacketSession>
{
public:
	PacketSession(class FSocket* Socket);
	~PacketSession();

	void Run();

	void Recv();

	void Disconnect();

public:
	class FSocket* Socket;

	TSharedPtr<class RecvWorker> RecvWorkerThread;
	TSharedPtr<class SendWorker> SendWorkerThread;

	// GameThread와 NetworkThread가 데이터 주고 받는 공용 큐.
	TQueue<TArray<uint8>> RecvPacketQueue;
};

 

 

#include "Network/PacketSession.h"
#include "NetworkWorker.h"
#include "Sockets.h"
#include "Common/TcpSocketBuilder.h"
#include "Serialization/ArrayWriter.h"
#include "SocketSubsystem.h"

PacketSession::PacketSession(class FSocket* Socket) : Socket(Socket)
{

}

PacketSession::~PacketSession()
{
	Disconnect();
}

void PacketSession::Run()
{
	RecvWorkerThread = MakeShared<RecvWorker>(Socket,AsShared());
}

void PacketSession::Recv()
{
}

void PacketSession::Disconnect()
{

}

 

 

TQueue<TArray<uint8>> RecvPacketQueue;

• GameThread와 NetworkThread가 데이터 주고 받는 공용 큐.

RecvWorker에서 RecvPacketQueue의 존재를 알아야 Recv한 패킷을 Queue에 넣어줄 수 있다.
그러면 RecvPacketQueue를 사용할 수 있게 패킷세션을 전달해줘야 하는데
어떻게 전달해줘야 할까?

그냥 레퍼런스나 포인터로 넘겨주게 되면 Session이 사라졌을 때 RecvWorker가 오염된 메모리를 참조하게 된다.
그래서 패킷세션의 shared_ptr를 RecvWorker에게 전달해주고, 전달 받은 패킷세션의 shared_ptr을 weak_ptr로 저장하고 있다.

근데 왜 전달 받은 패킷세션의 shared_ptr을 weak_ptr로 저장하고 있을까?
바로 shared_ptr의 고질적인 문제인 순환참조 때문이다.
그래서 자식 클래스에서 부모 클래스를 참조할 때 weak_ptr로 받으면 생명주기에는 영향을 주지 않기 떄문에
weak_ptr로 받아주었다.

 

 

RecvWorkerThread = MakeShared<RecvWorker>(Socket,AsShared());

• AsShared() : 서버 작업을 할 때 사용하던 C++ 표준 shared_from_this() 함수라고 보면 된다.

 

Smart Pointers in Unreal Engine

---