2.1 多线程编程基础

为了更平稳地过渡，在真正进入UE的多线程渲染知识之前，先学习或重温一下多线程编程的基础知识。

2.1.1 多线程概述

多线程（Multithread）编程的思想早在单核时代就已经出现了，当时的操作系统（如Windows95）就已经支持多任务的功能，其原理就是在单核中切换不同的上下文（Context），以便每个进程中的线程都有时间获得执行指令的机会。

但到了2005年，当单核主频接近4GHz时，CPU硬件厂商英特尔和AMD发现，速度也会遇到自己的极限：那就是单纯的主频提升，已经无法明显提升系统整体性能。

随着单核计算频率摩尔定律的缓慢终结，Intel率先于2005年发布了奔腾D和奔腾四至尊版840系列，首次支持了两个物理级别的线程计算单元。此后十多年，多核CPU得到蓬勃发展，由AMD制造的Ryzen 3990X处理器已经拥有64个核心128个逻辑线程。

锐龙（Ryzen）3990X的宣传海报中赫然凸显的核心与线程数量。

硬件的多核发展，给软件极大的发挥空间。应用程序可以充分发挥多核多线程的计算资源，各个应用领域由此也产生多线程编程模型和技术。作为游戏的发动机Unreal Engine等商业引擎，同样可以利用多线程技术，以便更加充分地提升效率和效果。

使用多线程并发带来的作用总结起来主要有两点：

• 分离关注点。通过将相关的代码与无关的代码分离，可以使程序更容易理解和测试，从而减少出错的可能性。比如，游戏引擎中通常将文件加载、网络传输放入独立的线程中，既可以不阻碍主线程，也可以分离逻辑代码，使得更加清晰可扩展。
• 提升性能。人多力量大，这样的道理同样用到CPU上（核多力量大）。相同量级的任务，如果能够分散到多个CPU中同时运行，必然会带来效率的提升。

但是，随着CPU核心数量的提升，计算机获得的效益并非直线提升，而是遵循Amdahl's law（阿姆达尔定律），Amdahl's law的公式定义如下：

公式的各个分量含义如下：

• \(S_{latency}\)：整个任务在多线程处理中理论上获得的加速比。
• \(s\)：用于执行任务并行部分的硬件资源的线程数量。
• \(p\)：可并行处理的任务占比。

举个具体的栗子，假设有8核16线程的CPU用于处理某个任务，这个任务有70%的部分是可以并行处理的，那么它的理论加速比为：

由此可见，多线程编程带来的效益并非跟核心数呈直线正比，实际上它的曲线如下所示：

阿姆达尔定律揭示的核心数和加速比图例。由此可见，可并行的任务占比越低，加速比获得的效果越差：当可并行任务占比为50%时，16核已经基本达到加速比天花板，无论后面增加多少核心数量，都无济于事；如果可并行任务占比为95%时，到2048个核心才会达到加速比天花板。

虽然阿姆达尔定律给我们带来了残酷的现实，但是，如果我们能够提升任务并行占比到接近100%，则加速比天花板可以得到极大提升：

如上公式所示，当\(p=1\)（即可并行的任务占比100%）时，理论上的加速比和核心数量成线性正比！！

举个具体的例子，在编译Unreal Engine工程源码或Shader时，由于它们基本是100%的并行占比，理论上可以获得接近线性关系的加速比，在多核系统中将极大地缩短编译时间。

利用多线程并发提高性能的方式有两种：

• 任务并行（task parallelism）。将一个单个任务分成几部分，且各自并行运行，从而降低总运行时间。这种方式虽然看起来很简单直观，但实际操作中可能会很复杂，因为在各个部分之间可能存在着依赖。
• 数据并行（data parallelism）。任务并行的是算法（执行指令）部分，即每个线程执行的指令不一样；而数据并行是指令相同，但执行的数据不一样。SIMD也是数据并行的一种方式。

上面阐述了多线程并发的益处，接下来说说它的副作用。总结起来，副作用如下：

• 导致数据竞争。多线程访问常常会交叉执行同一段代码，或者操作同一个资源，又或者多核CPU的高度缓存同步问题，由此变化带来各种数据不同步或数据读写错误，由此产生了各种各样的异常结果，这便是数据竞争。
• 逻辑复杂化，难以调试。由于多线程的并发方式不唯一，不可预知，所以为了避免数据竞争，常常加入复杂多样的同步操作，代码也会变得离散、片段化、繁琐、难以理解，增加代码的辅助，对后续的维护、扩展都带来不可估量的阻碍。也会引发小概率事件难以重现的BUG，给调试和查错增加了数量级的难度。
• 不一定能够提升效益。多线程技术用得到确实会带来效率的提升，但并非绝对，常和物理核心、同步机制、运行时状态、并发占比等等因素相关，在某些极端情况，或者用得不够妥当，可能反而会降低程序效率。

2.1.2 多线程概念

本小节将阐述多线程编程技术中常涉及的基本概念。

• 进程（Process）

进程（Process）是操作系统执行应用程序的基本单元和实体，它本身只是个容器，通常包含内核对象、地址空间、统计信息和若干线程。它本身并不真正执行代码指令，而是交由进程内的线程执行。

对Windows而言，操作系统在创建进程时，同时也会给它创建一个线程，该线程被称为主线程（Primary thread, Main thread）。

对Unix而言，进程和主线程其实是同一个东西，操作系统并不知道有线程的存在，线程更接近于lightweight processes（轻量级进程）的概念。

进程有优先级概念，Windows下由低到高为：低（Low）、低于正常（Below normal）、正常（Normal）、高于正常（Above normal）、高（High）、实时（Real time）。（见下图）

默认情况下，进程的优先级为Normal。优先级高的进程将会优先获得执行机会和时间。

• 线程（Thread）

线程（Thread）是可以执行代码的实体，通常不能独立存在，需要依附在某个进程内部。一个进程可以拥有多个线程，这些线程可以共享进程的数据，以便并行或并发地执行多个任务。

在单核CPU中，操作系统（如Windows）可能会采用轮循（Round robin）的方式进行调度，使得多个线程看起来是同时运行的。（下图）

在多核CPU中，线程可能会安排在不同的CPU核心同时运行，从而达到并行处理的目的。

采用SMP的Windows在多核CPU的执行示意图。等待处理的线程被安排到不同的CPU核心。

每个线程可拥有自己的执行指令上下文（如Windows的IP（指令寄存器地址）和SP（栈起始寄存器地址））、执行栈和TLS（Thread Local Storage，线程局部缓存）。

Windows线程创建和初始化示意图。

线程局部存储（Thread Local Storage）是一种存储持续期，对象的生命周期与线程一样，在线程开始时分配，线程结束时回收。每个线程有该对象自己的实例，访问和修改这样的对象不会造成竞争条件（Race Condition）。

线程也存在优先级概念，优先级越高的将优先获得执行指令的机会。

线程的状态一般有运行状态、暂停状态等。Windows可用以下接口切换线程状态：

// 暂停线程
DWORD SuspendThread(HANDLE hThread);
// 继续运行线程
DWORD ResumeThread(HANDLE hThread);

#include <threads.h>
thread_local int foo = 0;

// Visual C++, Intel C/C++ (Windows systems), C++Builder, Digital Mars C++
__declspec(thread) int number;

// Solaris Studio C/C++, IBM XL C/C++, GNU C, Clang, Intel C++ Compiler (Linux systems)
__thread int number;

// C++ Builder
int __thread number;

int a = 10;
volatile int *p = &a;
int b, c;
b = *p;
c = *p;

int cnt = 0;
auto f = [&]{cnt++;};
std::thread t1{f}, t2{f}, t3{f};

std::atomic<int> cnt{0};	// 给cnt加入原子操作。
auto f = [&]{cnt++;};
std::thread t1{f}, t2{f}, t3{f};

#include <iostream>
#include <thread>
#include <chrono>

void foo()
{
    // simulate expensive operation
    std::this_thread::sleep_for(std::chrono::seconds(1));
}
 
int main()
{
    std::cout << "starting thread...\n";
    std::thread t(foo); // 构造线程对象，且传入被执行的函数。
 
    std::cout << "waiting for thread to finish..." << std::endl;
    t.join(); // 加入主线程，使得主线程必须等待该线程执行完毕。
 
    std::cout << "done!\n";
}

starting thread...
waiting for thread to finish...
done!

#include <vector>
#include <thread>
#include <future>
#include <numeric>
#include <iostream>
 
void accumulate(std::vector<int>::iterator first,
                std::vector<int>::iterator last,
                std::promise<int> accumulate_promise)
{
    int sum = std::accumulate(first, last, 0);
    accumulate_promise.set_value(sum);  // Notify future
}
 
int main()
{
    // Demonstrate using promise<int> to transmit a result between threads.
    std::vector<int> numbers = { 1, 2, 3, 4, 5, 6 };
    std::promise<int> accumulate_promise;
    std::future<int> accumulate_future = accumulate_promise.get_future();
    std::thread work_thread(accumulate, numbers.begin(), numbers.end(),
                            std::move(accumulate_promise));
 
    // future::get() will wait until the future has a valid result and retrieves it.
    // Calling wait() before get() is not needed
    //accumulate_future.wait();  // wait for result
    std::cout << "result = " << accumulate_future.get() << '\n';
    work_thread.join();  // wait for thread completion
}

result = 21

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <future>
#include <string>
#include <mutex>
 
std::mutex m;
struct X {
    void foo(int i, const std::string& str) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << str << ' ' << i << '\n';
    }
    void bar(const std::string& str) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << str << '\n';
    }
    int operator()(int i) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << i << '\n';
        return i + 10;
    }
};
 
template <typename RandomIt>
int parallel_sum(RandomIt beg, RandomIt end)
{
    auto len = end - beg;
    if (len < 1000)
        return std::accumulate(beg, end, 0);
 
    RandomIt mid = beg + len/2;
    auto handle = std::async(std::launch::async,
                             parallel_sum<RandomIt>, mid, end);
    int sum = parallel_sum(beg, mid);
    return sum + handle.get();
}
 
int main()
{
    std::vector<int> v(10000, 1);
    std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << '\n';
 
    X x;
    // Calls (&x)->foo(42, "Hello") with default policy:
    // may print "Hello 42" concurrently or defer execution
    auto a1 = std::async(&X::foo, &x, 42, "Hello");
    // Calls x.bar("world!") with deferred policy
    // prints "world!" when a2.get() or a2.wait() is called
    auto a2 = std::async(std::launch::deferred, &X::bar, x, "world!");
    // Calls X()(43); with async policy
    // prints "43" concurrently
    auto a3 = std::async(std::launch::async, X(), 43);
    a2.wait();                     // prints "world!"
    std::cout << a3.get() << '\n'; // prints "53"
} // if a1 is not done at this point, destructor of a1 prints "Hello 42" here

The sum is 10000
43
Hello 42
world!
53

#include <atomic>
#include <future>
#include <iostream>

template<typename T>
struct node
{
    T data;
    node* next;
    node(const T& data) : data(data), next(nullptr) {}
};
 
template<typename T>
class stack
{
 public:
    std::atomic<node<T>*> head;	// 堆栈头, 采用原子操作.
 public:
    // 入栈操作
    void push(const T& data)
    {
        node<T>* new_node = new node<T>(data);
 
        // 将原有的头指针作为新节点的下一节点.
        new_node->next = head.load(std::memory_order_relaxed);
 
        // 将新的节点和老的头部节点做对比测试, 如果new_node->next==head, 说明其它线程没有修改head, 可以将head替换成new_node, 从而完成push操作.
        // 反之, 如果new_node->next!=head, 说明其它线程修改了head, 将其它线程修改的head保存到new_node->next, 继续循环检测.
        while(!head.compare_exchange_weak(new_node->next, new_node,
                                        std::memory_order_release,
                                        std::memory_order_relaxed))
            ; // 空循环体
    }
};

int main()
{
    stack<int> s;
    
    auto r1 = std::async(std::launch::async, &stack<int>::push, &s, 1);
    auto r2 = std::async(std::launch::async, &stack<int>::push, &s, 2);
    auto r3 = std::async(std::launch::async, &stack<int>::push, &s, 3);
    
    r1.wait();
    r2.wait();
    r3.wait();
    
    // print the stack's values
    node<int>* node = s.head.load(std::memory_order_relaxed);
    while(node)
    {
        std::cout << node->data << " ";
        node = node->next;
    }
}

2 3 1

#include <iostream>
#include <map>
#include <string>
#include <chrono>
#include <thread>
#include <mutex>
 
std::map<std::string, std::string> g_pages;
std::mutex g_pages_mutex;	// 声明互斥量
 
void save_page(const std::string &url)
{
    // simulate a long page fetch
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::string result = "fake content";
 	
    // 配合std::lock_guard使用, 可以及时进入和释放互斥量.
    std::lock_guard<std::mutex> guard(g_pages_mutex);
    g_pages[url] = result;
}
 
int main() 
{
    std::thread t1(save_page, "http://foo");
    std::thread t2(save_page, "http://bar");
    t1.join();
    t2.join();
 
    // safe to access g_pages without lock now, as the threads are joined
    for (const auto &pair : g_pages) {
        std::cout << pair.first << " => " << pair.second << '\n';
    }
}

http://bar => fake content
http://foo => fake content

#include <chrono>
#include <thread>
#include <mutex>

bool flag;
std::mutex m;

void wait_for_flag()
{
    std::unique_lock<std::mutex> lk(m); // 这里采用std::unique_lock而非std::lock_guard. std::unique_lock可以实现尝试获得锁, 如果当前以及被其它线程锁定, 则延迟直到其它线程释放, 然后才获得锁.
    while(!flag)
    {
        lk.unlock(); // 解锁互斥量
        std::this_thread::sleep_for(std::chrono::milliseconds(100));  // 休眠100ms，在此期间，其它线程可以进入互斥量，以便更改flag标记。
        lk.lock();   // 再锁互斥量
    }
}

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>
 
std::mutex m;
std::condition_variable cv;	// 声明条件变量
std::string data;
bool ready = false;
bool processed = false;
 
void worker_thread()
{
    // 等待直到主线程改变ready为true.
    std::unique_lock<std::mutex> lk(m);
    cv.wait(lk, []{return ready;});
 
    // 获得了互斥量的锁
    std::cout << "Worker thread is processing data\n";
    data += " after processing";
 
    // 发送数据给主线程
    processed = true;
    std::cout << "Worker thread signals data processing completed\n";
 
    // 手动解锁, 以便主线程获得锁.
    lk.unlock();
    cv.notify_one();
}
 
int main()
{
    std::thread worker(worker_thread);
 
    data = "Example data";
    // send data to the worker thread
    {
        std::lock_guard<std::mutex> lk(m);
        ready = true;
        std::cout << "main() signals data ready for processing\n";
    }
    cv.notify_one();
 
    // wait for the worker
    {
        std::unique_lock<std::mutex> lk(m);
        cv.wait(lk, []{return processed;});
    }
    std::cout << "Back in main(), data = " << data << '\n';
 
    worker.join();
}

main() signals data ready for processing
Worker thread is processing data
Worker thread signals data processing completed
Back in main(), data = Example data after processing

#include <iostream>
#include <future>
#include <thread>
 
int main()
{
    // 从packaged_task获取的future
    std::packaged_task<int()> task([]{ return 7; }); // wrap the function
    std::future<int> f1 = task.get_future();  // get a future
    std::thread t(std::move(task)); // launch on a thread
 
    // 从async()获取的future
    std::future<int> f2 = std::async(std::launch::async, []{ return 8; });
 
    // 从promise获取的future
    std::promise<int> p;
    std::future<int> f3 = p.get_future();
    std::thread( [&p]{ p.set_value_at_thread_exit(9); }).detach();
 	
    // 等待所有future
    std::cout << "Waiting..." << std::flush;
    f1.wait();
    f2.wait();
    f3.wait();
    std::cout << "Done!\nResults are: " << f1.get() << ' ' << f2.get() << ' ' << f3.get() << '\n';
    t.join();
}

Waiting...Done!
Results are: 7 8 9