UE5多线程｜ThreadPool

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

这是侑虎科技第1917篇文章，感谢作者南京周润发供稿。欢迎转发分享，未经作者授权请勿转载。如果您有任何独到的见解或者发现也欢迎联系我们，一起探讨。（QQ群：793972859）

作者主页：

https://www.zhihu.com/people/xu-chen-71-65

当有持续时间短，又比较杂的异步任务时，可以使用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.类型定义

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

1. 线程池

FQueuedThreadPool：线程池基类，定义了线程池的接口。

Allocate：创建线程池，类型为FQueuedThreadPoolBase。

Create：创建若干工作线程。

AddQueuedWork：向线程池添加任务。

RetractQueuedWork：撤回任务。

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

FQueuedThreadPool有多种实现：

• FQueuedThreadPoolBase

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

成员：

FThreadPoolPriorityQueue QueuedWork：待处理任务的队列。

TArray QueuedThreads：等待接收任务的空闲线程。

TArray AllThreads：所有工作线程。

FCriticalSection* SynchQueue：保护任务队列的CriticalSection，因为任务队列会被多线程修改。

• FQueuedLowLevelThreadPool

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

• FQueuedThreadPoolWrapper

• FQueuedThreadPoolDynamicWrapper

• FQueuedThreadPoolTaskGraphWrapper

2. 线程池线程

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

成员：

DoWorkEvent：通知线程有任务要执行的Event。

QueuedWork：当前线程正在执行的Work。

Thread：Runnable对应的线程。

函数：

Run：主函数，可认为是一个等待、执行任务的循环。

DoWork：由ThreadPool调用，传入一个任务并执行。

3. 任务

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 WorkPool; Work集合。

TMap ScheduledWork; 当前正在被执行的Work。

std::atomic MaxConcurrency; 最多允许多少Work在后台线程池中运行。

std::atomic CurrentConcurrency; 当前在后台线程池中运行的Work。

• FScheduledWork

成员中出现了FScheduledWork类型，它是一个容器，存储了真正的IQueuedWork，同时也是IQueuedWork的子类，有DoThreadedWork接口。

其中128行执行了异步任务，131行通知FQueuedThreadPoolWrapper任务执行完，可调度下个任务，会在下面介绍。

• 初始化

构造函数如下，主要接受一个线程池作为后台线程池，InMaxConcurrency表示最多同时在后台线程池中执行多少个任务。

• AddQueuedWork

AddQueuedWork首先把任务加到QueuedWork中，然后执行Schedule函数，默认参数为空。

Schedule函数最重要的是下面几行。首先从QueuedWork中获取要执行的任务，然后递增CurrentConcurrency。接着通过AllocateWork获取一个FScheduledWork对象，并把InnerWork封装在里面，然后把FScheduledWork交给后台线程池运行。

WorkPool容器就缓存了已创建的FScheduledWork对象，AllocateWork会首先从中获取，没有再创建，避免性能上的浪费。

• 执行

FScheduledWork执行完DoThreadedWork后，会调用Release，继续让线程池执行剩余任务，并把自己重置，加入WorkPool中，等待下次使用。

图示如下：

文末，再次感谢南京周润发 的分享， 作者主页：https://www.zhihu.com/people/xu-chen-71-65， 如果您有任何独到的见解或者发现也欢迎联系我们，一起探讨。（QQ群： 793972859 ）。

近期精彩回顾