UE5多线程｜ThreadPool

作者：admin
/
时间：2小时前
/
浏览：13 次
/
分类：厚积薄发

【USparkle专栏】如果你深怀绝技，爱“搞点研究”，乐于分享也博采众长，我们期待你的加入，让智慧的火花碰撞交织，让知识的传递生生不息！

当有持续时间短，又比较杂的异步任务时，可以使用ThreadPool，用固定数量的工作线程执行任务，不每次都创建新线程。UE4和UE5的线程池有很大区别，UE4线程池会真的创建很多线程，而UE5主要线程池底层复用了TaskGraph的线程，线程池只是逻辑上的概念。

一、创建线程池

线程池在FEngineLoop::PreInitPreStartupScreen函数中创建。

GThreadPool

类型为FQueuedLowLevelThreadPool，是UE5中的新实现，线程数量由FPlatformMisc::NumberOfWorkerThreadsToSpawn()确定。

GIOThreadPool

类型为FQueuedThreadPool，线程数量由FPlatformMisc::NumberOfIOWorkerThreadsToSpawn()确定，Client为4，Server为2。

GBackgroundPriorityThreadPool

类型为FQueuedThreadPool，Client为2，Server为1。

GLargeThreadPool

类型为FQueuedLowLevelThreadPool，数量由FPlatformMisc::NumberOfCoresIncludingHyperthreads()确定。

二、使用线程池

虽然线程池实现比Runnable复杂，但使用方式也比较简单。

1. Async函数
最常见用法，Async函数可设置EAsyncExecution::ThreadPool参数，指定任务在ThreadPool里执行。

函数内部会创建TAsyncQueuedWork封装Function和Promise，然后使用AddQueuedWork接口把任务加到GThreadPool中。

AddQueuedWork是线程池最重要的接口。

2. AsyncPool函数
与Async类似，但可以指定线程池和Work优先级。

3. 手动调用AddQueuedWork
AddQueuedWork函数只需要接受IQueuedWork作为参数，TAsyncQueuedWork只是一个子类，我们可以创建子类，做自定义操作，这样也能指定使用哪个线程池。

比如引擎中Encode LightMap的操作，就使用了FAsyncEncode类：

三、线程池实现

1.类型定义
类型定义可分为线程池，线程池线程，任务。

线程池
FQueuedThreadPool：线程池基类，定义了线程池的接口。
Allocate：创建线程池，类型为FQueuedThreadPoolBase。
Create：创建若干工作线程。
AddQueuedWork：向线程池添加任务。
RetractQueuedWork：撤回任务。

AddQueuedWork和RetractQueuedWork是线程池提供给外部调用的主要接口，注意会在多线程中被调用。

FQueuedThreadPool有多种实现：

FQueuedThreadPoolBase

最常用，线程池的基础实现，GIOThreadPool和GBackgroundPriorityThreadPool都会使用。

成员：
FThreadPoolPriorityQueue QueuedWork：待处理任务的队列。
TArray<FQueuedThread*> QueuedThreads：等待接收任务的空闲线程。
TArray<FQueuedThread*> AllThreads：所有工作线程。
FCriticalSection* SynchQueue：保护任务队列的CriticalSection，因为任务队列会被多线程修改。

FQueuedLowLevelThreadPool

底层线程使用TaskGraph的ThreadPool，UE5中GThreadPool的默认实现。

FQueuedThreadPoolWrapper
FQueuedThreadPoolDynamicWrapper
FQueuedThreadPoolTaskGraphWrapper

线程池线程
FQueuedThread：继承自FRunnable，表示线程池中的工作线程。可以想象，它大部分时间都处于idle状态，当有任务来时才工作。

成员：
DoWorkEvent：通知线程有任务要执行的Event。
QueuedWork：当前线程正在执行的Work。
Thread：Runnable对应的线程。

函数：
Run：主函数，可认为是一个等待、执行任务的循环。
DoWork：由ThreadPool调用，传入一个任务并执行。

任务
IQueuedWork：可排队任务的基类接口，供线程池使用。

接口：
DoThreadedWork：执行任务。

IQueuedWork有多种实现：

TAsyncQueuedWork

最常用，Async和AsyncPool函数中使用。

DoThreadedWork：通过SetPromise执行任务。

FAsyncTaskBase

可操作内容更多。

DoThreadedWork：通过Task执行任务。

类图如下：

常用部分已高亮显示

2. FQueuedThreadPoolBase

线程池创建

FQueuedThreadPoolBase是默认线程池，FQueuedThreadPool::Allocate函数中构造。

线程池通过Create函数初始化，主要工作是创建InNumQueuedThreads数量的工作线程，使用FQueuedThread类封装，并把创建的线程加入QueuedThreads和AllThreads容器中，QueuedThreads中存储了当前线程池中处于空闲状态的线程。还要创建CriticalSection对象SynchQueue，用于保护对QueuedWork和QueuedThreads的访问。

FQueuedThread
FQueuedThread继承自FRunnable，是一个可运行任务的抽象，其Create函数如下。首先创建DoWorkEvent，用于做多线程同步，然后创建一个底层的Thread。线程创建好后进入Run方法，初始没有任务，线程在DoWorkEvent上等待，处于休眠状态。

添加任务

观察AddQueuedWork函数，添加任务时分成了两种情况。

如果线程池中尚有空闲线程，即下图中的情况1，QueuedThreads中有元素，那么把任务分配给其中一个线程即可，这里还有一个细节，QueuedThreads采用栈管理，先进后出，这可以更好利用CPU Cache，因为这个Thread可能刚运行过，同时也可以避免数组中的元素移动。得到Thread后，调用DoWork方法添加任务。

另一种情况是所有线程都在忙碌，QueuedThreads中没有元素，这时只能把InQueuedWork暂存到QueuedWork中，等线程执行完之前任务后再做处理。

FQueuedThread::DoWork方法用于通知一个Thread要执行任务了，首先把InQueuedWork设置到其QueuedWork属性上，然后执行DoWorkEvent的Trigger方法，唤醒该Thread。注意这里加了一个MemoryBarrier，是为了避免CPU指令乱序优化导致1071行在1074行之后执行，导致错误。

执行任务

执行任务通过属性的Run函数实现。Thread一开始会在DoWorkEvent上等待，被DoWork函数唤醒后，会获取之前被赋值的QueuedWork，执行DoThreadedWork函数，这里是真正执行任务。执行完成后再调用ThreadPool的ReturnToPoolOrGetNextJob函数，尝试获取暂存的QueuedWork并执行，若没有就把Thread归还到QueuedThreads中，之后在DoWorkEvent上等待，进入休眠状态。

流程图示：

3. TAsyncQueuedWork
线程池中的任务，包装了一个Function对象，DoThreadWork函数中使用给Promise SetValue的形式来执行Function。

以上就是UE线程池常用的FQueuedThreadPoolBase，FQueuedThread，TAsyncQueuedWork组合。

以下内容是UE5的改动。

4. FQueuedLowLevelThreadPool
在UE5中，非Editor模式下GThreadPool实现变成了FQueuedLowLevelThreadPool。底层使用了TaskGraph，相关内容放在后面看，这里只分析与线程池相关的部分。

UE希望把多线程操作尽量放在TaskGraph里，这样好管理。CPU物理核心数量是有限的，如果TaskGraph和ThreadPool都创建了核心数量的线程，其实在各自管理，两边线程都跑满就会产生更多的CPU调度开销。

Create

其实不需要Create了，因为自己不创建线程，初始化在构造函数里完成，主要任务是获取LowLevelTasks::FScheduler单例。

FQueuedThreadPool::Create只是实现一下纯虚函数。

LowLevelTasks::Fscheduler管理了TaskGraph中的Workers线程，包括ForegroundWorkers和BackgroundWorkers，向Worker线程分发任务，细节后面再看。

5. AddQueuedWork

首先创建FQueuedWorkInternalData对象来存储QueuedWork相关数据，然后设置到InQueuedWork.InternalData属性。

FQueuedWorkInternalData类包装了一个LowLevelTasks::FTask，FTask用于把QueuedWork包装成TaskGraph里可执行的东西。Retract函数用于取消任务，但线程池场景下不需要考虑取消。

Task.Init函数调用有点绕，464行先把InQueuedWork包装成一个Lambda函数，然后在Init实现里面再把Lambda包装到另一个TFunction里面。这样就把InQueuedWork存到Task里面了，往后操作只和TaskGraph有关，与线程池无关了。

FScheduler::TryLaunch把Task添加到任务队列中，等待Worker线程来消费。

6. 执行任务
TaskGraph中Worker线程的Run函数会循环获取任务执行，细节放后面TashGraph里看，这里只看一个调用栈。

下图中1的位置是Worker线程取Task，2的位置是执行InQueuedWork->DoThreadedWork()，终于又回到了线程池。

总体来看，FQueuedLowLevelThreadPool其实就是TaskGraph，和Async函数中传EAsyncExecution::TaskGraph是一个效果。

7. FQueuedThreadPoolWrapper
不是真正的线程池，而是另一个线程池的包装，任务都会转发过去。UE5 Editor下GThreadPool就会设置成这个，包装了GLargeThreadPool，目的为共用GLargeThreadPool中的线程，类似FQueuedLowLevelThreadPool共用TaskGraph的线程，因为Editor下后台任务更多，因此单独使用了GLargeThreadPool。这么做的目的还是减少线程创建。

主要成员

FQueuedThreadPool* WrappedQueuedThreadPool; 包装的ThreadPool。
TArray<FScheduledWork*> WorkPool; Work集合。
TMap<IQueuedWork*, FScheduledWork*> ScheduledWork; 当前正在被执行的Work。
std::atomic MaxConcurrency; 最多允许多少Work在后台线程池中运行。
std::atomic CurrentConcurrency; 当前在后台线程池中运行的Work。