程序丨如何在任务系统中实现无锁的工作窃取（二）？
译者：王磊（未来的未来）
审校：崔国军（飞扬971）
正如在这个系列的上一篇文章（点此阅读）中所承诺的那样，今天我们将会看到如何在我们的任务系统中分配任务的时候去摆脱创建任务和删除任务的问题。只要我们愿意为此牺牲一些内存，任务的分配就可以以更有效的方式来处理。由此产生的性能改善是巨大的，所以这么做当然是值得的。
为什么使用new操作和delete操作速度会很慢
如前所述，new操作基本上是在后台调用一个通用的分配器。 这个分配器必须满足小，中和（非常）大的分配需求。此外，new操作是保证线程安全的，使用像互斥体或临界区等同步原语来保护可变的共享数据。
因此，在我们这样一个特殊情况下，我们应该能够找到比new操作和delete操作性能更好的方法就不足为奇了。 尽管通用分配器的性能和特性不断得到改进，但它们永远无法击败定制的分配器。
还有一点要记住，使用new操作分配实例会迫使我们调用这些实例的delete操作，以避免内存泄漏。 这反过来又限制了我们把任务的删除推迟到一个帧的结尾，要把它们存储在一个辅助数组中。另外，我们在Finish（）函数中还需要另一个原子计数器。
一个内存池分配器？
我们的任务系统只做Job类型的分配，所以使用内存池的allocator / free list听起来像是一个完美的选择。使用（无锁）内存池分配器确实提高了性能，但仍不能解决需要推迟删除任务实例的问题。
一个专门的分配器
要认识到这个问题的关键是，我们每一帧都一遍又一遍地做同样的事情。我们生成并分配N个任务，并在一帧的结尾删除它们所有N个任务。那么为什么不去掉分配任务和释放任务这两个事情呢？
这可以通过使用一个预先分配的Job实例全局数组来轻松完成。因为我们的Job结构是一个POD类型，所以我们不需要担心调用构造函数或是析构函数的问题。我们可以在初始化任务系统的时候，分配一个比如说有4096个任务的数组，使用该数组作为进行响应分配请求的环形缓冲区，并在完全关闭任务系统的时候，比如想像是应用程序退出的情况下，释放这个数组。
在这里我们所需要做的全部内容是一个全局的任务数组，以及一个原子计数器，它可以作为一种无锁环形缓冲区的分配器，如果你想这样做的话。 Allocate（）函数可以简单地变成下面这样：
static Job g_jobAllocator[MAX_JOB_COUNT];
static uint32_t g_allocatedJobs = 0u;
Job* AllocateJob(void)
｛
const uint32_t index = atomic::Increment(&g_allocatedJobs);
return &g_jobAllocator[(index-1) % MAX_JOB_COUNT];
｝
正如你可能已经知道的那样，在MAX_JOB_COUNT是2的幂的情况下，模数表达式（index-1）％MAX_JOB_COUNT可以变成二进制与运算：
return &g_jobAllocator[(index-1u) & (MAX_JOB_COUNT-1u)];
可以看出，原子计数器是一个单调递增的整数，永远不会被重置，因此我们使用模运算访问全局的g_jobAllocator数组，并将其高效地转换为环形缓冲区。
在生产环境的代码中，您应该确保采取其他措施，确保在单个框架中不会分配超过MAX_JOB_COUNT个任务。此外，你可以memset（）Job实例（或时间戳），以确保代码不会触及过去帧的任务。
用这种方法实现的一个重要的优点就是我们不再需要在一个框架内删除任务。 这意味着我们可以去掉全局辅助数组和相应的原子计数器，从而导致对Finish（）函数的简化实现：
void Finish(Job* job)
｛
const int32_t unfinishedJobs = atomic::Decrement(&job->unfinishedJobs);
if ((unfinishedJobs == 0) && (job->parent))
｛
Finish(job->parent);
｝
｝
这个方法所需的预分配内存的数量在比较大的内存需求中可以忽略不计。 每帧有4096个任务，数组需要4096 * 64 = 256 KB，这在今天的平台上不算什么。
尽管这个实现在性能方面比原来的方法有一个很好的提升，但是我们仍然可以做的更好。 读者应该知道接下来会发生什么。
使用线程本地化（thread-local）
那么，我们怎么能比每个分配使用一个原子操作做的更好呢？ 当然，答案只有根本不使用原子操作！ 原子操作比互斥和类似的方法的开销要低得多，但它们也不是免费的。
通过将计数器和任务的预分配数组移动到线程本地来存储，我们不再需要任何原子操作来从环形缓冲区中分配任务：
Job* AllocateJob(void)
｛
const uint32_t index = g_allocatedJobs++;
return &g_jobAllocator[index & (MAX_JOB_COUNT-1u)];
｝
在上面的代码示例中，g_allocatedJobs和g_jobAllocator是在线程本地存储上分配的。 注意没有任何原子操作或是昂贵的函数调用。所以开销已经尽可能的低了。
还有一点需要注意的是，与我们以前的方法相比，这种方法需要更多的内存，因为所有的工作线程现在都需要一个预分配的MAX_JOB_COUNT大小的Job实例数组。 在8个工作线程的情况下，这会将内存需求从256KB提升到2MB，至少在8核心的机器上，我仍然认为这是少量的内存需求。
性能
同样，性能是在主频为3.4 GHz的IntelCore i7-2600K CPU上测得的，具有4个超线程（=8个逻辑内核）的物理内核。
随着实现的改进，现在的运行时间如下：
Old | New | Perf. increase | |
Single jobs | 18.5 ms | 9.9 ms | 1.86x |
parallel_for | 5.3 ms | 1.35 ms | 3.92x |
虽然只是一些相当小的变化，但是这真得到了巨大的性能提升！
展望
下一次，我们将要解决任务系统的问题：实施无锁任务队列。
【版权声明】
原文作者未做权利声明，视为共享知识产权进入公共领域，自动获得授权。
原文链接：https://blog.molecular-matters.com/2015/09/08/job-system-2-0-lock-free-work-stealing-part-2-a-specialized-allocator/
今日推荐
一键添加
加小编微信，享双重福利
1.加入GAD程序猿交流群，获取行业干货；
2.领取60G腾讯内部分享等独家程序资料。