【ue4】多线程之FRunnable
【ue4】多线程之FRunnable
前言
今日起开始研究ue4的多线程机制。感觉这部分内容还挺多的，所以想从根源上弄清楚一些，这样以后看ue4就是单线程的了，岂不爽哉。大略读了一下源码，又结合了网上一些大牛的文章，基本的学习路线差不多确定了，这里列一下。
基本的多线程机制 -- FRunnable + FRunnableThread 线程池 -- FQueuedThread + FQueuedThreadPool 异步任务 -- FAsyncTask TaskGraph
本篇先从最简单的FRunnable开始。
FRunnable
FRunnable类的声明位于Source\Runtime\Core\Public\HAL\Runnable.h，没有实现。
其中HAL的意思是Hardware Abstraction Layer，即硬件抽象层。位于HAL中的文件大部分都是平台无关的，即在隐藏了平台差异化的接口，类似于RHI的概念。
class CORE_API FRunnable
{
public:
virtual bool Init() { return true; }
virtual uint32 Run() = 0;
virtual void Stop() { }
virtual void Exit() { }
virtual class FSingleThreadRunnable* GetSingleThreadInterface( )
{
return nullptr;
}
virtual ~FRunnable() { }
};没错，以上便是FRunnable的全部实现，有够简单。其中GetSingleThreadInterface()可以先忽略，后面会讲到。
显然，FRunnable类并非真正的多线程类，它没有任何一个成员变量，只有几个流程向的虚函数，明眼人一瞧便知上层业务逻辑便是子类化该类，并实现这些虚函数，即可在多线程的某些阶段被调用到，所以称FRunnable为线程执行体。
那么我们下一步要弄清楚的便是这几个虚函数到底是在哪个地方被调用的，尤其是纯虚函数Run()。
FRunnable
FRunnableThread
FRunnableThread类的声明位于Source\Runtime\Core\Public\HAL\RunnableThread.h。
FRunnableThread类的实现位于Source\Runtime\Core\Private\HAL\ThreadingBase.cpp。
从文件坐落的位置可以看出，这个类也是平台无关的。
class CORE_API FRunnableThread
{
protected:
// core data =========================
uint32 ThreadID;
FString ThreadName;
EThreadPriority ThreadPriority;
FRunnable* Runnable;
FEvent* ThreadInitSyncEvent;
uint64 ThreadAffinityMask;
// core data =========================
public:
// create function ====================
static FRunnableThread* Create(
class FRunnable* InRunnable,
const TCHAR* ThreadName,
uint32 InStackSize = 0,
EThreadPriority InThreadPri = TPri_Normal,
uint64 InThreadAffinityMask = FPlatformAffinity::GetNoAffinityMask() );
protected:
virtual bool CreateInternal( FRunnable* InRunnable, const TCHAR* InThreadName,
uint32 InStackSize = 0,
EThreadPriority InThreadPri = TPri_Normal, uint64 InThreadAffinityMask = 0 ) = 0;
// create function ====================
public:
// virtual function ====================
virtual void SetThreadPriority( EThreadPriority NewPriority ) = 0;
virtual void Suspend( bool bShouldPause = true ) = 0;
virtual bool Kill( bool bShouldWait = true ) = 0;
virtual void WaitForCompletion() = 0;
virtual ~FRunnableThread();
private:
virtual void Tick() {}
// virtual function ====================
private:
// about tls =========================
static uint32 RunnableTlsSlot;
public:
static uint32 GetTlsSlot();
protected:
TArray<FTlsAutoCleanup*> TlsInstances;
void SetTls();
void FreeTls();
static FRunnableThread* GetRunnableThread()
{
FRunnableThread* RunnableThread = (FRunnableThread*)FPlatformTLS::GetTlsValue( RunnableTlsSlot );
return RunnableThread;
}
// about tls =========================
};其中有一部分数据是与TLS(Thread Local Storage, 线程本地存储)相关的，这部分我们再讲到TLS时再来分解，这里视而不见就好。
那么FRunnableThread类剩下的内容大概有三个部分。
其一是其核心数据层，ID、Name、Priority显而易见是描述事物不可或缺的东西。其中最重要的是这个类包含了一个FRunnable类型的成员变量，那么自然就像上面讲的那样，通过控制该成员变量的一些虚函数的调用，来控制上层逻辑在该类中的执行流程。
其二便是一些虚函数和纯虚函数了，这部分函数大部分都是对外的接口，是用户调用以控制该线程流程的外部接口。之所以是虚函数和纯虚函数，其大部分原因自然是因为不同的平台可能有不同的对线程控制的底层api调用，所以必须将该线程类子类化到某个具体的平台下，实现这些平台相关的接口，才能真正被用户使用。
其三便是真正创建线程的静态函数Create()，这是我们需要重点来看的地方。
FRunnableThread* FRunnableThread::Create(
class FRunnable* InRunnable,
const TCHAR* ThreadName,
uint32 InStackSize,
EThreadPriority InThreadPri,
uint64 InThreadAffinityMask)
{
FRunnableThread* NewThread = nullptr;
if (FPlatformProcess::SupportsMultithreading())
{
check(InRunnable);
NewThread = FPlatformProcess::CreateRunnableThread();
if (NewThread)
{
if (NewThread->CreateInternal(InRunnable,ThreadName,InStackSize,InThreadPri,InThreadAffinityMask) == false)
{
delete NewThread;
NewThread = nullptr;
}
}
}
else if (InRunnable->GetSingleThreadInterface())
{
NewThread = new FFakeThread();
if (NewThread->CreateInternal(InRunnable,ThreadName,InStackSize,InThreadPri) == false)
{
delete NewThread;
NewThread = nullptr;
}
}
return NewThread;
}可以看到，在该函数里首先执行FPlatformProcess::SupportsMultithreading()来判断当前平台是否支持多线程，如果不支持多线程，则逻辑会走到else分支里。不支持多线程的情况我们会在后文详细展开，所以目前先只把精力集中于支持多线程的情况即可。
支持多线程的逻辑里，大致分两步。首先调用FPlatformProcess::CreateRunnableThread()创建一个平台相关的FRunnableThread*子类，然后调用该子类实现的CreateInternal()纯虚函数。
我们下面先来看看此处的与跨平台Platform相关的接口是如何实现具体化到某个平台的，然后再看下几个平台上FRunnableThread子类的具体实现。
Platform跨平台
与跨平台Platform相关的文件位于SourceWork\Engine\Engine\Source\Runtime\Core\Public\HAL目录下，形如PlatformXXX.h，如PlatformProcess.h、PlatformAffinity.h等。
PlatformXXX.h
在PlatformXXX.h文件中，会根据宏定义确定的当前处于哪个平台，来选择include哪个具体的nnnPlatformXXX.h文件，其中nnn为平台名称(Windows、Android、Apple等)。如PlatFormProcess.h的内容如下。
#pragma once
#include "CoreTypes.h"
#if PLATFORM_WINDOWS
#include "Windows/WindowsPlatformProcess.h"
#elif PLATFORM_PS4
#include "PS4/PS4Process.h"
#elif PLATFORM_XBOXONE
#include "XboxOne/XboxOneProcess.h"
#elif PLATFORM_MAC
#include "Mac/MacPlatformProcess.h"
#elif PLATFORM_IOS
#include "IOS/IOSPlatformProcess.h"
#elif PLATFORM_LUMIN
#include "Lumin/LuminPlatformProcess.h"
#elif PLATFORM_ANDROID
#include "Android/AndroidProcess.h"
#elif PLATFORM_QUAIL
#include "Quail/QuailPlatformProcess.h"
#elif PLATFORM_HTML5
#include "HTML5/HTML5PlatformProcess.h"
#elif PLATFORM_LINUX
#include "Linux/LinuxPlatformProcess.h"
#elif PLATFORM_SWITCH
#include "Switch/SwitchPlatformProcess.h"
#endif而在位于Source\Runtime\Core\Public\nnn文件夹下具体的nnnPlatformXXX.h文件中，则是子类化了FGenericPlatformXXX类的FnnnPlatformXXX类，如下为WindowsPlatformProcess.h中的子类声明。
struct CORE_API FWindowsPlatformProcess : public FGenericPlatformProcess
{
//...
}其中FGenericPlatformXXX位于Source\Runtime\Core\Public\GenericPlatform\GenericPlatformXXX.h中，是抽象出平台无关接口的地方，由该类的子类具体实现某平台的具体方法。
且在nnnPlatformXXX.h中，往往还会用typedef关键字给所有的子类函数声明为同一个别名，这样我们在调用时就不必再次使用宏定义来决定选择哪个子类了，其形往往如下所示。
typedef FnnnPlatformXXX FPlatformXXX;如WindowsPlatformProcess.h里便有如下声明。
typedef FWindowsPlatformProcess FPlatformProcess;让我们回到上面FRunnableThread的Create()函数里，再次关注使用FPlatformProcess的地方，便可以清楚地理解为调用平台相关的子类方法了。
NewThread = FPlatformProcess::CreateRunnableThread();比如上面的调用在Windows平台下，就会返回一个该平台下的FRunnableThread子类。
FRunnableThread* FWindowsPlatformProcess::CreateRunnableThread()
{
return new FRunnableThreadWin();
}自然后面调用的CreateInternal()函数，也是该子类具体实现的平台相关的函数。
FRunnableThread
至此，关于FRunnable和FRunnableThread的核心代码我们已经参详完毕。但是我们却发现了一件事情，那就是FRunnable的那些虚函数，好像还并没有看到调用它们的地方。由此可见，一定是在具体平台的FRunnableThread子类里面来调用的。所以下面我们就来看几个具有代表性的子类。
FRunnableThreadWin
FRunnableThreadWin位于Source\Runtime\Core\Private\Windows\WindowsRunnableThread.h。
class FRunnableThreadWin : public FRunnableThread
{
HANDLE Thread;
static void SetThreadName( uint32 ThreadID, LPCSTR ThreadName )
{
//...
}
static ::DWORD STDCALL _ThreadProc( LPVOID pThis )
{
check(pThis);
return ((FRunnableThreadWin*)pThis)->GuardedRun();
}
uint32 GuardedRun();
uint32 Run();
public:
FRunnableThreadWin( ) : Thread(NULL) {}
~FRunnableThreadWin( )
{
if (Thread != NULL)
{
Kill(true);
}
}
virtual void SetThreadPriority( EThreadPriority NewPriority ) override
{
ThreadPriority = NewPriority;
::SetThreadPriority(Thread, TranslateThreadPriority(ThreadPriority));
}
virtual void Suspend( bool bShouldPause = true ) override
{
check(Thread);
if (bShouldPause == true)
{
SuspendThread(Thread);
}
else
{
ResumeThread(Thread);
}
}
virtual bool Kill( bool bShouldWait = false ) override
{
check(Thread && "Did you forget to call Create()?");
bool bDidExitOK = true;
if (Runnable)
{
Runnable->Stop();
}
if (bShouldWait == true)
{
WaitForSingleObject(Thread,INFINITE);
}
CloseHandle(Thread);
Thread = NULL;
return bDidExitOK;
}
virtual void WaitForCompletion( ) override
{
WaitForSingleObject(Thread,INFINITE);
}
protected:
virtual bool CreateInternal( FRunnable* InRunnable, const TCHAR* InThreadName,
uint32 InStackSize = 0,
EThreadPriority InThreadPri = TPri_Normal, uint64 InThreadAffinityMask = 0 ) override
{
check(InRunnable);
Runnable = InRunnable;
ThreadAffinityMask = InThreadAffinityMask;
ThreadInitSyncEvent = FPlatformProcess::GetSynchEventFromPool(true);
// create the new thread
{
Thread = CreateThread(NULL, InStackSize, _ThreadProc, this, STACK_SIZE_PARAM_IS_A_RESERVATION | CREATE_SUSPENDED, (::DWORD *)&ThreadID);
}
if (Thread == NULL)
{
Runnable = nullptr;
}
else
{
FThreadManager::Get().AddThread(ThreadID, this);
ResumeThread(Thread);
ThreadInitSyncEvent->Wait(INFINITE);
ThreadName = InThreadName ? InThreadName : TEXT("Unnamed UE4");
SetThreadPriority(InThreadPri);
}
// cleanup the sync event
FPlatformProcess::ReturnSynchEventToPool(ThreadInitSyncEvent);
ThreadInitSyncEvent = nullptr;
return Thread != NULL;
}
};
FRunnableThreadWin实现了父类的纯虚函数CreateInternal()。在该函数里调用了WINAPI来真正在Windows平台创建线程，即CreateThread()函数，这个函数会创建一个线程，创建完成后将执行传入的回调函数_ThreadProc()。而在_ThreadProc()里则执行了成员函数GuardedRun()，由此可见，该线程的实际入口函数是GuaradedRun()。
uint32 FRunnableThreadWin::GuardedRun()
{
uint32 ExitCode = 0;
FPlatformProcess::SetThreadAffinityMask(ThreadAffinityMask);
if (FPlatformMisc::IsDebuggerPresent() && !GAlwaysReportCrash)
{
ExitCode = Run();
}
else
{
__try
{
ExitCode = Run();
}
__except (ReportCrash( GetExceptionInformation() ))
{
GWarn->Flush();
// crashed.
ExitCode = 1;
GError->HandleError();
FPlatformMisc::RequestExit( true );
}
}
return ExitCode;
}可以看到，GuardedRun()函数本身也没干啥，只不过是在调用另一个成员函数Run()，且在调用失败的时候捕获异常而已，Guarded本身的中文意思也是保守的、安全的，这样看来Run()才是真正的入口函数。
uint32 FRunnableThreadWin::Run()
{
uint32 ExitCode = 1;
check(Runnable);
if (Runnable->Init() == true)
{
ThreadInitSyncEvent->Trigger();
SetTls();
ExitCode = Runnable->Run();
Runnable->Exit();
FreeTls();
}
else
{
ThreadInitSyncEvent->Trigger();
}
return ExitCode;
}终于，我们看到了线程执行体FRunnable的那些虚函数的调用了，几乎全在Run()函数里，先是Init()，然后再执行Run()，最后执行Exit()，只剩下FRunnable的Stop()函数在FRunnableThreadWin的成员函数Kill()里调用了。
由此也可以看出，FRunnable的Run()函数在一次多线程回调里只会执行一次。
FRunnableThreadPThread
FRunnableThreadPThread位于Source\Runtime\Core\Private\HAL\PThreadRunnableThread.h。
由此可以它并非一个具体化到平台的子类，而是进一步将FRunnableThread的一些过程又进一步封装或者固定了一下，进一步提取一些各个平台都公有的流程。
首先，它固定了Run()函数的实现，即将线程执行体FRunnable的执行流程固定了下来，其实现与FRunnableThreadWin保持一致，参见上文即可。
其次，线程回调函数_ThreadProc()不再调用GuardedRun()，而是直接调用Run()，并在其前后分别调用虚函数PreRun()和PostRun()以便子类拓展。
static void *STDCALL _ThreadProc(void *pThis)
{
check(pThis);
FRunnableThreadPThread* ThisThread = (FRunnableThreadPThread*)pThis;
ThisThread->ThreadID = FPlatformTLS::GetCurrentThreadId();
FThreadManager::Get().AddThread(ThisThread->ThreadID, ThisThread);
FPlatformProcess::SetThreadAffinityMask(ThisThread->ThreadAffinityMask);
// run the thread!
ThisThread->PreRun();
ThisThread->Run();
ThisThread->PostRun();
pthread_exit(NULL);
return NULL;
}继承FRunnableThreadPThread的类如下。
FRunnableThreadAndorid FRunnableThreadApple FRunnableThreadUnix
FRunnableThread子类
FFakeThread
在上节的分析中，我们从一开始便忽略了不支持多线程的情况(见上节一开始)，那么现在我们可以来看一下不支持多线程的情况了，即在FRunnableThread函数里FPlatformProcess::SupportsMultithreading()的返回值为false的情况。
FRunnableThread* FRunnableThread::Create(
class FRunnable* InRunnable,
const TCHAR* ThreadName,
uint32 InStackSize,
EThreadPriority InThreadPri,
uint64 InThreadAffinityMask)
{
FRunnableThread* NewThread = nullptr;
if (FPlatformProcess::SupportsMultithreading())
{
//...
}
else if (InRunnable->GetSingleThreadInterface())
{
NewThread = new FFakeThread();
if (NewThread->CreateInternal(InRunnable,ThreadName,InStackSize,InThreadPri) == false)
{
delete NewThread;
NewThread = nullptr;
}
}
return NewThread;
}另一条分支一开始的情况便是调用线程执行体里我们唯一没有分析过的虚函数GetSingleThreadInterface()来判断是否可以获取FSingleThreadRunnable。
FSingleThreadRunnable位于Source\Runtime\Core\Public\Misc\SingleThreadRunnable.h。
其中Misc的意思是(Miscellaneous, 杂项)，作者没有想好放哪的东西一般都会放在这里。。有一种不受待见的感觉。我们不妨亲切地称FSingleThreadRunnable为单线程执行体，以下便是它的所有实现。
class CORE_API FSingleThreadRunnable
{
public:
virtual ~FSingleThreadRunnable() { }
/* Tick function. */
virtual void Tick() = 0;
};在不支持多线程的情况下，如果传入的线程执行体FRunnable如果可以通过GetSingleThreadInterface返回单线程执行体FSingleThreadRunnable的话，便会创建一个类型为FFakeThread的类，以在单线程的情况下模拟多线程的机制。
FFakeThread位于Source\Runtime\Core\Private\HAL\ThreadingBase.cpp，对外是不可见的。
FFakeThread与我们上面介绍的FRunnableThreadWin一样，都是FRunnableThread的子类。
class FFakeThread : public FRunnableThread
{
static uint32 ThreadIdCounter;
bool bIsSuspended;
FSingleThreadRunnable* Runnable;
public:
FFakeThread() : bIsSuspended(false), Runnable(nullptr)
{
ThreadID = ThreadIdCounter++;
FThreadManager::Get().AddThread(ThreadID, this);
}
virtual ~FFakeThread()
{
FThreadManager::Get().RemoveThread(this);
}
// tick
virtual void Tick() override
{
if (Runnable && !bIsSuspended)
{
Runnable->Tick();
}
}
public:
virtual void SetThreadPriority(EThreadPriority NewPriority) override
{
}
virtual void Suspend(bool bShouldPause) override
{
bIsSuspended = bShouldPause;
}
virtual bool Kill(bool bShouldWait) override
{
FThreadManager::Get().RemoveThread(this);
return true;
}
virtual void WaitForCompletion() override
{
FThreadManager::Get().RemoveThread(this);
}
// create internal
virtual bool CreateInternal(FRunnable* InRunnable, const TCHAR* InThreadName,
uint32 InStackSize,
EThreadPriority InThreadPri, uint64 InThreadAffinityMask) override
{
Runnable = InRunnable->GetSingleThreadInterface();
if (Runnable)
{
InRunnable->Init();
}
return Runnable != nullptr;
}
};可以看到，FFakeThread与其他子类有几处明显的不同。
首先，FFakeThread的成员变量Runnable的类型为单线程执行体FSingleThreadRunnable。
其次，观察其CreateInternal()函数，可以发现，该类并没有真正创建某平台相关的线程，而只是从传进的线程执行体获取单线程执行体，并赋值给其成员变量，在成功赋值的同时调用线程执行体的Init()虚函数。
而其单线程执行体FSingleThreadRunnable的虚函数Tick()则在FFakeThread的虚函数Tick()里执行。
所以，我们只需要子类化一个FSingleThreadRunnable，并在其Tick()虚函数里实现自己的逻辑，该逻辑便会随着FFakeThread的Tick()而每帧调用，分享当前线程的Tick时间片。
但是问题又来了，目前为止，我们并没有见到FFakeThread的Tick()函数是在哪里触发的。这就跟接下来要看的全局单例类FThreadManager有关了。
FFakeThread
Demo
如果我们想要一段上层逻辑既可以在多线程下执行，又可以在不支持多线程的情况下分时间片执行，那么就可以声明一个既继承自FRunnable，同时又继承自FSingleThreadRunnable的类。
重写GetSingleThreadInterface()函数，返回this指针即可。
重写Run()函数，执行多线程时的逻辑。
重写Tick()函数，执行单线程时在每帧的时间片里执行的逻辑。
class FCustomRunnable: public FRunnable, public FSingleThreadRunnable
{
public:
virtual class FSingleThreadRunnable* GetSingleThreadInterface() override { return this; }
uint32 Run() override
{
while(CanRun())
{
RunOneFrame()
}
}
void Tick() override
{
RunOneFrame()
}
private:
void RunOneFrame()
{
//...
}
}FThreadManager
FThreadManager位于Source\Runtime\Core\Public\HAL\ThreadManager.h。
FThreadManager的实现极其简单，如下所示。
class CORE_API FThreadManager
{
TMap<uint32, class FRunnableThread*> Threads;
FCriticalSection ThreadsCritical;
public:
void AddThread(uint32 ThreadId, class FRunnableThread* Thread);
void RemoveThread(class FRunnableThread* Thread);
void Tick();
const FString& GetThreadName(uint32 ThreadId);
static FThreadManager& Get();
};FThreadManager是一个全局单例，它维护了一个全局唯一的TMap，用来存储所有的FRunnableThread，并提供Add()、Remove()、GetName()这些方法用来修改和查询该TMap。
其Tick()函数如下所示。
void FThreadManager::Tick()
{
if (!FPlatformProcess::SupportsMultithreading())
{
FScopeLock ThreadsLock(&ThreadsCritical);
// tick all registered threads.
for (TPair<uint32, FRunnableThread*>& ThreadPair : Threads)
{
ThreadPair.Value->Tick();
}
}
}在不支持多线程的情况下，FThreadManager的Tick()函数里会执行所有FRunnableThread的Tick()函数。
所以，当我们将FFakeThread类型的对象添加进该全局管理器里时，该全局管理器的Tick()便接管了FFakeThread的Tick()，用来在不支持多线程的情况下模拟多线程的机制。
从上面FFakeThread的实现里可以看到，在FFakeThread的构造函数里调用了AddThread()函数将其加入全局管理器，在FFakeThread的析构函数里，以及其Kill()函数、WaitForCompletion()函数里，都会调用RemoveThread()函数将其从全局管理器中移除。
而FThreadManager的Tick()函数，则是在游戏主Tick之后执行，如下代码所示。
void FPreLoadScreenManager::GameLogicFrameTick()
{
//...
FTicker::GetCoreTicker().Tick(DeltaTime);
FThreadManager::Get().Tick();
}另外，其他FRunnableThread的子类对象，其添加进全局管理器以及从全局管理器中删除的时机与FFakeThread都大同小异，有的可能是在CreateInternal()时加进去的，有的可能是在_ThreadProc()回调时加进去的，可参考具体代码查看。
FThreadManager
Demo
讲了这么多，下面来看一个使用FRunnable和FRunnableThread来创建多线程的最简实例吧。
class FMyRunnable : public FRunnable
{
public:
FMyRunnable() : TimeToDie(false) {}
~FMyRunnable() {}
virtual uint32 Run() override
{
while (!TimeToDie.Load(EMemoryOrder::Relaxed))
{
FPlatformMisc::MemoryBarrier();
DoWorkOneTime();
FPlatformProcess::Sleep(0.033f);
}
}
virtual void Stop()
{
TimeToDie = true;
}
private:
void DoWorkOneTime()
{
//...
}
TAtomic<bool> TimeToDie;
}
// ==========================================
FRunnableThread* Thread = nullptr;
// start thread work
void Start()
{
FMyRunnable* MyRunnable = new FMyRunnable();
Thread = FRunnableThread::Create(MyRunnable, TEXT("MyThread"), 0, TPri_BelowNormal);
}
// stop thread work
void Stop()
{
Thread->Kill()
Thread->WaitForCompletion();
delete Thread;
}后记
FRunnableThread类图
个人认为FRunnable和FRunnableThread是ue4多线程的核心地基，后续的线程池、异步任务、TaskGraph等机制，都是在这个地基的基础上进一步搭建起来的。所以对这部分代码作一次总的分解是很有必要的。
无奈本人能力有限，中间省略了不少细枝末节，这里把能想到的关键词记一下，后面有机会了再详细补充一下。
FEvent TAtomic原子变量与内存模型 线程的挂起和恢复 TLS(Thread Local Storage, 线程本地存储) 多线程同步问题 无锁数据结构的实现
个人认为对于多线程而言，最需要弄清楚的还是游荡在各个线程中的共享资源的同步问题，打算后面把ue4多线程的机制有一定了解之后，可以再看一下ue4是怎么解决各种资源同步问题的，以及互斥锁、信号量，无锁队列等的实现细节，都有不少的地方可以再进一步弄清楚一些。